Examples and Cookbook

Examples and Cookbook

This page provides fully realized, production-ready FML reference implementations. Each example includes complete code and an architectural walkthrough.

---

Example 1: Web Search, Iterate & Synthesis

A complete pipeline that searches the web, iterates over extracted URLs to scrape each page independently, and synthesizes a final executive brief.

system("You are an automated research analyst.")

parameter("search_term", type=string)
parameter("limit", type=int, default=3)

require mcp web_scraper

components {
    schema("PageMeta") {
        url: string
        title: string
    }
}

# Session 1: Gather relevant links using built-in search
session("gather_links", target="links") {
    use search
    + Search the web for: {{ .params.search_term }}
    - Extract the top {{ .params.limit }} URLs found.
    schema $PageMeta[]
}

# Session 2: Iterate over gathered links and scrape each page
session("scrape_pages", target="scrapes", after="gather_links", iterate=context.links) {
    use mcp web_scraper
    + Call scrape_page for the URL: {{ .it.url }}
    - Summarize the main findings from the page content.
    schema {
        url: string
        summary: string # Core findings from the page
    }
}

# Session 3: Synthesize all scraped pages into a single brief
session("synthesize", after="scrape_pages") {
    context "Scrapes: {{ .context.scrapes | json }}"
    - Compile the summaries into a final research brief.
    schema {
        brief: string # Consolidated report
        key_themes: string[]
        sources: string[]
    }
}

Architectural Walkthrough

1. Search Tool Usage: Uses use search inside Session 1 without declaring require search at the root.
2. Reusable Component: PageMeta is defined in components and referenced as $PageMeta[].
3. Iterative Processing: Session 2 declares iterate=context.links and after="gather_links". The session loops over each link, accessing {{ .it.url }}.
4. Mandatory Array Output: Because Session 2 uses iterate, its schema automatically collects each iteration's output into an array.
5. Context Serialization: Session 3 loads the array of summaries using {{ .context.scrapes | json }} and generates the final synthesized brief.

---

Example 2: Call Output Routing & Database Synchronization

Demonstrates deterministic data piping between tools, inserting records into a collection, and querying the database without LLM intervention.

require mcp github
require collection internal_db

parameter("org", type=string)

# Step 1: Deterministic call to fetch repositories
call("list_org_repos") -> vars:raw_repos {
    org = "{{ .params.org }}"
}

# Step 2: Deterministic call to insert records directly into DB
call("internal_db_insert") {
    table   = "repositories"
    records = $(vars.raw_repos)
}

# Step 3: LLM session to query the collection and structure active projects
session("query_db", target="selected") {
    use collection internal_db

    + Query the database table 'repositories' to find active projects
      with more than 50 stars and commits in the last 30 days.

    - Organize the matching repositories into the output schema.

    schema {
        name: string
        stars: int
        description: string
    }[]
}

Architectural Walkthrough

1. Tool Piping via vars: list_org_repos routes its output to vars:raw_repos.
2. Type Preservation: internal_db_insert consumes $(vars.raw_repos) as a native array without string serialization.
3. No LLM Overhead: The initial fetch and database population happen before any session executes, consuming zero LLM tokens.

---

Example 3: Security Incident Triage (Token Optimization & Scripted IIFE)

Demonstrates how to clean and filter 500+ raw logs inside an IIFE PreCall so the LLM only processes critical anomalies.

system(`You are a lead security operations engineer.`)

parameter("service_id", type="string")
parameter("environment", type="string", default="production", enum=staging|production)

require mcp cloudwatch

session("triage_incident", target="incident_report") {
    use mcp cloudwatch { allowlist = ["query_metrics"] }

    # Step 1: PreCall to fetch raw logs
    call("fetch_logs") -> vars:raw_logs {
        service = "{{ .params.service_id }}"
        env     = "{{ .params.environment }}"
        limit   = 500
    }

    # Step 2: PreCall IIFE to sanitize, filter, and deduplicate
    call("filter_critical_logs") -> vars:filtered_logs {
        logs = $(vars.raw_logs)
        code(
            (
                (() => {
                    const entries = args.logs || [];
                    const critical = entries.filter(e => e.level === "CRITICAL" || e.level === "FATAL");
                    
                    # Deduplicate by error signature
                    const seen = new Set();
                    const deduplicated = [];
                    for (const entry of critical) {
                        if (!seen.has(entry.error_signature)) {
                            seen.add(entry.error_signature);
                            deduplicated.push({
                                timestamp: entry.timestamp,
                                signature: entry.error_signature,
                                message: entry.message
                            });
                        }
                    }
                    return deduplicated;
                })()
            )
        )
    }

    # Step 3: Pre-prompt (Free-form Chain-of-Thought reasoning)
    + Inspect the critical error signatures:
      {{ .vars.filtered_logs | json }}
      Query CloudWatch metrics for anomaly spikes during these timestamps.
      Formulate an initial root-cause hypothesis.

    # Step 4: Prompt (Strict Schema Output)
    - Output the final incident triage report.

    schema {
        service: string
        severity: low|medium|high|critical
        incident_summary: string
        affected_components: string[]
        recommended_mitigation: string
    }
}

Architectural Walkthrough

1. Token Savings: 500 raw log entries (~250 KB) are reduced to a deduplicated array of unique signatures (~5 KB) before reaching the LLM prompt.
2. Chain-of-Thought via Pre-Prompt: In Step 3, the model queries additional metrics and reasons through the error signatures freely before being constrained by the output schema in Step 4.

---

Example 4: Conditional Execution & Cascading Skips

Demonstrates the expect attribute for executing remediation actions only when specific risk thresholds are met.

parameter("repo_url", type=string)

require mcp security_scanner
require mcp slack_notifier

session("scan_repository", target="scan_result") {
    use mcp security_scanner

    + Scan the repository at {{ .params.repo_url }} for secrets and vulnerabilities.
    - Output the scan score and vulnerability count.

    schema {
        score: int # 0 to 100
        critical_count: int
        is_compromised: bool
    }
}

# This session ONLY runs if critical vulnerabilities are detected!
session("alert_security_team", after="scan_repository", expect="context.scan_result.critical_count > 0") {
    use mcp slack_notifier

    + Send urgent alert to #security-ops regarding repository {{ .params.repo_url }}.
      Vulnerabilities detected: {{ .context.scan_result.critical_count }}.

    - Confirm alert delivery.

    schema {
        alert_sent: bool
        channel: string
        timestamp: string
    }
}

# If alert_security_team is skipped, this session is automatically skipped as well!
session("create_jira_ticket", after="alert_security_team") {
    - Create ticket for tracking.
    schema { ticket_id: string }
}

Architectural Walkthrough

1. Conditional Evaluation: If critical_count == 0, expect="context.scan_result.critical_count > 0" evaluates to false.
2. Cascading Skip: alert_security_team is skipped. Because create_jira_ticket depends on alert_security_team via after, it is automatically skipped as well.