ltpmcp

📜 3 tools

LTP — Lean Task Protocol

A structured plan language for AI agents. Compiled by LLM, executed by runtime.

LTP is a lightweight DSL (Domain-Specific Language) that separates planning from execution. Instead of letting an LLM pilot a multi-turn conversation loop (ReAct), LTP compiles the entire task into a structured plan in a single LLM call, then executes it deterministically step-by-step.

ltpmcp is primarily a library, but it also ships an MCP server and a CLI. Most users import it into their own agent code. It additionally exposes 3 MCP tools via the ltpmcp-server console-script (parse_plan, validate_plan, visualize_plan — they parse/validate/visualize LTP text, they do not execute plans), a ltpmcp CLI (parse / validate / visualize), and an optional FastAPI app (ltpmcp-api, [api] extra). Note the distinction: the @TOOL names in plans below (@WEB_SEARCH, @EXEC_CODE, …) are plan instructions that the runtime routes to a tool_executor callback you supply — they are not the MCP tools above.

Installation

pip install mcpaisuite-ltpmcp

The only optional extra is [dev] (test tooling):

pip install mcpaisuite-ltpmcp[dev]   # pytest, pytest-asyncio, pytest-cov

Requirements: Python ≥ 3.11. Runtime dependencies: pydantic>=2.0, structlog>=24.0, click>=8.0, mcp>=1.0. Current version: 1.0.2. License: AGPL-3.0-or-later.

Quick Start

ltpmcp does not call an LLM or run tools by itself — you provide two async callbacks:

• llm_fn(messages) -> str — runs a chat completion (used for compilation and @LLM_* ops). messages is a list of {"role", "content"} dicts.
• tool_executor(tool_name, args, namespace) -> dict — executes a tool and returns {"success": bool, "output": ...} (and optionally "error").

import asyncio
from ltpmcp import LTPCompiler, LTPRuntime

async def my_llm(messages: list[dict]) -> str:
    # Call your model of choice; return the raw text completion.
    ...

async def my_tools(tool_name: str, args: dict, namespace: str) -> dict:
    # Route tool_name (e.g. "web_search", "execute_code") to your implementation.
    return {"success": True, "output": "..."}

async def main():
    # 1. Compile a goal into a plan (1 LLM call). Returns LTPPlan | None.
    compiler = LTPCompiler(llm_fn=my_llm)
    plan = await compiler.compile("What is the weather in Ibiza?", context="")
    if plan is None:
        raise RuntimeError("compilation failed")

    # 2. Execute the plan deterministically.
    runtime = LTPRuntime()
    result = await runtime.execute(plan, tool_executor=my_tools, llm_fn=my_llm)
    print(result["response"])

asyncio.run(main())

Verified signatures

# Exports (from ltpmcp import ...)
LTPPlan, LTPStep, LTPCondition, LTPParser, LTPRuntime, LTPCompiler, validate_plan, plan_to_mermaid,
PlanPolicy, PlanVerdict, Violation, verify_plan

# Compiler — 1 LLM call, Micro-CLI → LTP. Returns None on failure.
LTPCompiler(llm_fn).compile(goal: str, context: str = "") -> LTPPlan | None

# Runtime — deterministic step executor.
LTPRuntime(audit_fn=None).execute(
    plan,
    tool_executor,                 # async (tool_name, args, namespace) -> dict
    llm_fn,                        # async (messages) -> str
    agent_fn=None,                 # async (config_dict) -> str, for @AGENT_SOLVE
    namespace="default",
    max_replans=3,
    progress_callback=None,        # async (event_dict) -> None
) -> dict   # {"response", "variables", "steps_executed", "terminated"}

# Parser — raw LTP text -> plan (used internally by the compiler).
LTPParser.parse(raw: str) -> LTPPlan | None

# Static analyzer — returns a list of error/warning strings ([] = valid).
validate_plan(plan: LTPPlan) -> list[str]

# Visualization — returns a fenced ```mermaid block.
plan_to_mermaid(plan: LTPPlan) -> str

# Static security verifier — proves a compiled plan against a policy. Never executes.
verify_plan(plan: LTPPlan, policy: PlanPolicy | None = None) -> PlanVerdict
PlanPolicy(allowed_tools=None, denied_tools=set(), max_steps=None,
           no_egress_after_sensitive=True, no_escape_hatches=True,
           forbidden_arg_patterns=DEFAULT_FORBIDDEN_ARGS)
# PlanVerdict.ok: bool   ·   PlanVerdict.violations: list[Violation]   ·   bool(verdict) == verdict.ok

execute(...) always returns a dict with keys: response (str), variables (dict), steps_executed (int), terminated (bool).

Validating and visualizing a plan

from ltpmcp import validate_plan, plan_to_mermaid

errors = validate_plan(plan)        # e.g. ["[S3] Undefined variable $foo in @RESPOND args"]
if errors:
    print("plan issues:", errors)

print(plan_to_mermaid(plan))        # Mermaid flowchart for docs / debugging

Why LTP?

The Problem with ReAct Loops

Traditional agentic systems use a ReAct (Reason-Act) loop:

LLM → decides action → executes tool → sees result → decides next action → ...

This works for simple tasks but breaks down as complexity increases:

• Token waste: each turn sends the growing conversation history
• Drift: the LLM loses focus after 5-10 turns
• Hallucination: the LLM fabricates tool results or skips steps
• Unpredictability: the same query can produce wildly different execution paths
• Debugging: impossible to trace what went wrong in a 15-turn chain

The LTP Solution

User query → LLM compiles plan (1 call) → Runtime executes steps (deterministic)

• 1 LLM call to plan the entire task
• Deterministic execution: Python runtime, not LLM judgment
• Explicit data flow: $variables passed between steps
• Conditional logic: ?IF evaluated in code, not by LLM
• Auditable: every step has clear inputs and outputs

Advanced LTP Operators

@AGENT_CRITIQUE

Critiques output against specified criteria using an LLM micro-call:

S1: @EXEC_CODE (code="cat server.py", language="shell") > $code
S2: @AGENT_CRITIQUE ($code, criteria="security vulnerabilities, injection risks, auth issues") > $review
S3: @RESPOND ($review)

@CONSENSUS

Finds common ground between two different perspectives:

S1: @AGENT_SOLVE (goal="Research pros of microservices", agent="research") > $pro
S2: @AGENT_SOLVE (goal="Research cons of microservices", agent="research") > $con
S3: @CONSENSUS ($pro, $con, topic="balanced architecture recommendation") > $balanced
S4: @RESPOND ($balanced)

@SELF_REFINE

Iteratively improves output over N rounds:

S1: @LLM_GENERATE (context=$data, format="report") > $draft
S2: @SELF_REFINE ($draft, goal="make it more concise and actionable", rounds=3) > $final
S3: @RESPOND ($final)

Core Concepts

1. The Plan

An LTP plan is a sequence of steps between PLAN_START and PLAN_END:

PLAN_START
S1: @WEB_SEARCH (query="prime minister of France 2026") > $results
S2: @LLM_EXTRACT ($results, target="person_name_and_title") > $answer
S3: @RESPOND ($answer)
PLAN_END

Each step has:

• ID: S1, S2, S3 — sequential identifier
• Tool: @WEB_SEARCH, @LLM_EXTRACT, @RESPOND — the operation to execute
• Arguments: (query="...", target="...") — parameters for the tool
• Output variable: > $results — where to store the result
• Optional condition: ?IF ($var == "value") THEN — conditional execution

2. Variables and Data Flow

Variables are the backbone of LTP. Data flows explicitly between steps:

S1: @WEB_SEARCH (query="weather Paris") > $raw_results
S2: @LLM_EXTRACT ($raw_results, target="temperature") > $temp
S3: @WRITE_FILE (path="weather.txt", content=$temp)
S4: @RESPOND ("Temperature in Paris:", $temp)

• Variables are prefixed with $
• A step’s output is stored in its > $var
• Subsequent steps reference $var in their arguments
• The runtime resolves $var to its actual value before executing each step

3. Three Types of Steps

Deterministic Tools (no LLM)

These execute directly — fast, predictable, no token cost:

The @HOST_* and @SHELL_EXEC tools have direct Micro-CLI shorthands (host, hostread, hostlist, shell). As with every @TOOL, the runtime maps them to a tool name (e.g. @HOST_EXEC → host_exec) and hands them to your tool_executor — ltpmcp does not run host commands itself.

LLM Micro-Calls (1 focused prompt each)

These use an LLM but with a short, targeted prompt — not the full conversation:

Each @LLM_* operation is a single LLM call with a focused prompt (~50-100 tokens). The LLM receives only the data it needs, not the entire conversation history.

Adaptive Agent (@AGENT_SOLVE)

For problems that require exploration, trial-and-error, or debugging — a mini ReAct loop scoped to one sub-problem:

@AGENT_SOLVE (goal="Fix the performance bottleneck: $issue", agent="code", max_turns=5) > $fix

• Spawns a sub-agent with limited turns (3-5)
• The sub-agent can observe, react, and retry
• Scoped to a specific goal — doesn’t drift
• Returns a single result to the plan’s variable store

This is the hybrid aspect of LTP: deterministic structure + adaptive exploration where needed.

4. Flow Control

Conditions

S2: ?IF ($check contains "Error") THEN @INSTALL_PKG (language="python", packages=["numpy"])
S3: ?IF (NOT_EMPTY($critical_bugs)) THEN @RESPOND ("Critical bugs found!", $critical_bugs)
S4: ?IF ($error_rate > "5") THEN @FETCH_PAGE (url="https://webhook.com/rollback")
S5: ?IF (IS_EMPTY($data)) THEN TERMINATE ("No data available")

Operators: ==, !=, >, <, >=, <=, contains, NOT_EMPTY, IS_EMPTY

Conditions are evaluated by the runtime in Python — not by the LLM. This means:

• No token cost for condition evaluation
• Deterministic behavior
• No hallucinated conditions

Iteration (?FOREACH)

Process each item in a list without spawning an agent:

S1: @FETCH_PAGE (url="https://api.internal/tickets") > $tickets
S2: ?FOREACH ($ticket IN $tickets) THEN @LLM_CLASSIFY ($ticket, categories=["bug", "feature"]) > $classified
S3: ?IF ($classified contains "bug") THEN @RESPOND ("Bugs found:", $classified)

• $ticket is the iteration variable — set to each item in $tickets
• The tool runs once per item — results are collected into the output variable as a list
• Works with any tool: @LLM_*, @FETCH_PAGE, @EXEC_CODE, etc.
• If the source is a string, the runtime attempts JSON parsing; if that fails, it wraps it in a single-item list

This eliminates the need for @AGENT_SOLVE on simple batch processing tasks.

Error Handling (ON_FAIL)

Define step-level fallbacks when a tool fails:

S2: @FETCH_PAGE (url="https://api.internal/data") > $data ON_FAIL @RETRY(3)
S3: @EXEC_CODE (code="process($data)", language="python") > $result ON_FAIL GOTO S5
S4: @FETCH_PAGE (url="https://backup-api.internal/data") > $data ON_FAIL TERMINATE ("All APIs offline")

Three modes:

• ON_FAIL @RETRY(N) — retry the step up to N times with 1s delay between attempts
• ON_FAIL GOTO S5 — jump to a different step on failure (skip or use alternative path)
• ON_FAIL TERMINATE ("message") — stop the plan with an error message

Without ON_FAIL, a failed step stores the error in the output variable and continues — the next step can check with ?IF.

Terminate

Stops execution with a message:

S2: ?IF (IS_EMPTY($logs)) THEN TERMINATE ("No memory leaks detected in yesterday's logs.")

Mid-Plan Pivot (RE-PLAN)

When a tool result completely invalidates the remaining plan, RE-PLAN calls the compiler again with context from the variables collected so far:

S1: @WEB_SEARCH (query="Acme Corp pricing 2026") > $results
S2: ?IF ($results contains "company closed") THEN RE-PLAN ("Acme Corp no longer exists. $results. Find alternative company to compare.")
S3: @LLM_EXTRACT ($results, target="pricing") > $pricing
S4: @RESPOND ($pricing)

How it works:

1. Runtime detects RE-PLAN
2. Builds context from current $variables (what data has been collected so far)
3. Calls the compiler again (1 LLM call) with the reason + variable context
4. Executes the new plan with the existing variables carried over
5. Maximum 3 re-plans to prevent infinite loops

This is cheaper than @AGENT_SOLVE (1 compile call vs 5-15 ReAct turns) and smarter than ON_FAIL (adapts the strategy, not just retries).

Respond

Returns the final answer to the user:

S5: @RESPOND ("Analysis complete. Report saved to", $file_path)

If @RESPOND receives an empty value and there are more steps after it, the runtime skips it and continues — this handles cases where the LLM compiler forgot a ?IF guard.

5. Data Handling

Dot Notation for JSON Traversal

When a tool returns structured JSON, access nested fields directly without an @LLM_EXTRACT call:

S1: @FETCH_PAGE (url="https://api.weather.com/paris") > $weather
S2: @RESPOND ("Temperature:", $weather.current.temp, "°C")

Instead of wasting an LLM micro-call:

# OLD — wastes 1 LLM call
S2: @LLM_EXTRACT ($weather, target="temperature") > $temp

# NEW — zero LLM cost, instant
S2: @RESPOND ("Temperature:", $weather.current.temp)

The runtime resolves $var.field.subfield by traversing the object in Python. Variables are stored as native Python types (dicts, lists, ints) — not stringified. This means $weather.current.temp is a direct dict[key] lookup, with no JSON re-parsing overhead. If a variable is somehow a string, the runtime attempts JSON parsing as a fallback.

Type Casting

Enforce output types to prevent downstream errors:

S1: @EXEC_CODE (code="print(len(data))", language="python") > $count:int
S2: @WEB_SEARCH (query="top 10 movies") > $movies:list
S3: @LLM_EVALUATE ($data, condition="is valid?") > $valid:bool
S4: @FETCH_PAGE (url="https://api.example.com/data") > $payload:json

Supported types: int, float, list, bool, json

The runtime casts after storing the result:

• :int — int(float(value)), ignores non-numeric
• :float — float(value)
• :list — JSON parse if string, wrap in list if single value
• :bool — TRUE/YES/1 → True, everything else → False
• :json — JSON parse string to dict/list

The Compiler

The LTP Compiler is a single LLM call with a few-shot prompt (LTPCompiler.compile). The prompt contains:

1. A compact command reference (search, fetch, exec, host, the @LLM_*/agent ops, ?if, foreach, ON_FAIL, replan, terminate, type casts, dot notation)
2. The user’s goal, substituted into the prompt
3. A fixed set of worked examples (goal → Micro-CLI plan), including conditionals, foreach, dot-notation, and ON_FAIL

compile() returns an LTPPlan on success or None if the model output cannot be parsed into at least one step. An optional context string is injected into the prompt (it is also how RE-PLAN passes the collected variables back into a recompile).

The compiler uses a Micro-CLI shorthand that _micro_to_ltp automatically converts to the canonical LTP format before parsing:

# Micro-CLI (what the LLM writes)          # Canonical LTP (what the parser receives)
memory_stats > stats                        S1: @MEMORY_STATS > $stats
?if $stats.fact_count > 5 : respond "MANY"  S2: ?IF ($stats.fact_count > 5) THEN @RESPOND ("MANY")
?if $stats.fact_count <= 5 : respond "FEW"  S3: ?IF ($stats.fact_count <= 5) THEN @RESPOND ("FEW")

The Micro-CLI is simpler for LLMs to generate (no @ prefixes, no THEN, no parentheses). The _micro_to_ltp converter handles the translation automatically and wraps the result in a PLAN_START...PLAN_END block.

Tool name resolution: a known command verb (e.g. search, fetch, host) maps to its @TOOL via a fixed table; an unknown verb that contains an underscore (e.g. memory_stats) is passed through as a direct @MEMORY_STATS-style tool reference. At runtime, LTPRuntime then maps @TOOL names to your executor’s tool names (via an internal _TOOL_MAP, falling back to the lowercased name without the @).

Why This Works with Small Models

The compiler prompt is structured as a translation task — convert natural language to a simple DSL. This is much easier for LLMs than multi-turn reasoning:

• 8B models (llama3.1, qwen3) can generate valid LTP for simple-to-moderate tasks
• 70B models (llama3.3, qwen3-70b) handle complex multi-step plans with conditions
• Cloud models (Claude, GPT-4) generate optimal plans consistently

The few-shot examples teach the format by demonstration, not by instruction.

Parse Fallback

If converting the Micro-CLI output to canonical LTP fails to yield any steps, the compiler tries to parse the model’s raw output directly (in case the model already emitted canonical LTP). If neither path produces at least one step — or the LLM call raises — compile() logs a warning and returns None. Callers should always handle the None case.

The Runtime

The runtime is pure Python — no LLM, no network, no randomness:

for step in plan.steps:
    # 1. Check condition (if any)
    if step.condition and not step.condition.evaluate(variables):
        continue

    # 2. Resolve $variables in arguments
    resolved_args = resolve_vars(step.args, variables)

    # 3. Execute the tool
    if step.is_llm_op:
        result = await llm_micro_call(step.tool, resolved_args)
    elif step.is_agent:
        result = await run_sub_agent(resolved_args)
    elif step.is_respond:
        return build_response(resolved_args)
    else:
        result = await tool_executor(step.tool, resolved_args)

    # 4. Store result in variable
    if step.output_var:
        variables[step.output_var] = result

Output Cleaning

The runtime automatically extracts clean data from tool outputs:

• Sandbox results ({'stdout': '2026-04-28', ...}) → "2026-04-28"
• Web search results → titled list with URLs
• Webpage content → clean markdown

This means @LLM_EXTRACT receives clean text, not raw JSON/HTML.

Static Analysis (validate_plan)

Before executing a plan, run the static analyzer to catch structural problems. It is pure Python (no LLM) and returns a list[str] — an empty list means no issues found.

from ltpmcp import validate_plan

for problem in validate_plan(plan):
    print(problem)

The analyzer checks for:

validate_plan is advisory — it returns messages but does not raise. Decide for yourself whether to block execution on a non-empty result.

Visualization (plan_to_mermaid)

plan_to_mermaid(plan) renders a plan as a Mermaid graph TD flowchart, returned as a fenced ```mermaid code block string. It draws:

• Sequential edges S1 --> S2
• Conditional edges labelled IF $var op value
• @PARALLEL groups as a subgraph
• ON_FAIL GOTO as a dotted edge, and ON_FAIL TERMINATE as a dotted edge to a terminal node
• GOTO jumps

from ltpmcp import plan_to_mermaid

mermaid = plan_to_mermaid(plan)
# Paste into any Markdown viewer that supports Mermaid, or render in docs.

An empty plan yields a single empty["Empty plan"] node.

Provable plans (verify_plan)

Because LTP compiles the entire plan in one call before anything executes, the plan is a static artifact you can inspect — and prove against a policy — ahead of time. verify_plan(plan, policy) walks the compiled steps and returns a PlanVerdict; it never executes anything.

from ltpmcp import LTPCompiler, PlanPolicy, verify_plan

plan = await LTPCompiler(llm_fn=my_llm).compile(goal)

verdict = verify_plan(plan, PlanPolicy(
    denied_tools={"@HOST_EXEC", "@SHELL_EXEC"},
    max_steps=20,
))

if not verdict.ok:
    raise RuntimeError(f"plan refused: {verdict.reason()}")

# only now do you run it
result = await LTPRuntime().execute(plan, my_tools, my_llm)

PlanPolicy fields (all opt-in; an empty policy still enforces the default secret/sensitive-path argument patterns):

PlanVerdict carries ok: bool and violations: list[Violation] (each with step_id, rule, detail); bool(verdict) is verdict.ok, and verdict.reason() joins the violations into one string. This is the same verifier kernelmcp can enforce automatically before it runs an LTP plan (pass a PlanPolicy as plan_policy).

Parallel Execution

Steps that share a parallel_group run concurrently (via asyncio.gather). In the Micro-CLI / LTP source this is expressed with an @PARALLEL { ... } block, which the parser tags onto each enclosed step’s LTPStep.parallel_group field (par_0, par_1, …). The runtime resolves each parallel step’s variables, executes them together, then stores all their output variables before continuing to the next sequential step.

Examples by Complexity

Level 1: Simple (1-2 steps)

“hi”

PLAN_START
S1: @RESPOND ("Hello! How can I help you today?")
PLAN_END

“What date is it?”

PLAN_START
S1: @EXEC_CODE (code="from datetime import date; print(date.today())", language="python") > $date
S2: @RESPOND ("Today's date is", $date)
PLAN_END

“Who is Tom Holland?”

PLAN_START
S1: @WEB_SEARCH (query="Tom Holland actor") > $results
S2: @LLM_EXTRACT ($results, target="biography_summary") > $bio
S3: @RESPOND ($bio)
PLAN_END

Level 2: Moderate (3-5 steps, data flow)

“Check weather in Paris, save to file, tell me temperature”

PLAN_START
S1: @FETCH_PAGE (url="https://wttr.in/Paris?format=j1") > $weather
S2: @LLM_EXTRACT ($weather, target="temperature_celsius_and_conditions") > $temp
S3: @WRITE_FILE (path="weather.json", content=$weather)
S4: @RESPOND ("Saved to weather.json. Temperature:", $temp)
PLAN_END

“Check if numpy is installed, install if needed, compute mean”

PLAN_START
S1: @EXEC_CODE (code="import numpy; print('ok')", language="python") > $check
S2: ?IF ($check contains "Error") THEN @INSTALL_PKG (language="python", packages=["numpy"])
S3: @EXEC_CODE (code="import numpy; print(numpy.mean([1,2,3,4,5]))", language="python") > $result
S4: @RESPOND ("Mean:", $result)
PLAN_END

Level 3: Complex (5-8 steps, conditions, agents)

“Find why our API is slow, fix it, verify the fix”

PLAN_START
S1: @FETCH_PAGE (url="https://metrics.internal/api/latency") > $metrics
S2: @LLM_ANALYZE ($metrics, task="identify_performance_bottleneck") > $bottleneck
S3: @AGENT_SOLVE (goal="Fix: $bottleneck", agent="code", max_turns=5) > $fix
S4: @EXEC_CODE (code="curl -w '%{time_total}' https://api.internal/health", language="shell") > $new_latency
S5: @LLM_EVALUATE ($metrics, $new_latency, condition="latency improved by >20%") > $improved
S6: ?IF ($improved == "FALSE") THEN @AGENT_SOLVE (goal="Retry fix: $bottleneck", agent="code", max_turns=5) > $fix_v2
S7: @STORE_FACT (content="API bottleneck: $bottleneck. Fix: $fix", importance=0.9)
S8: @RESPOND ("Performance analysis complete.", $fix, $new_latency)
PLAN_END

“Pull tickets, classify, alert on critical bugs”

PLAN_START
S1: @FETCH_PAGE (url="https://api.hubspot.com/tickets?status=open") > $tickets
S2: @LLM_CLASSIFY ($tickets, categories=["feature_request", "bug", "critical_bug"]) > $classified
S3: @LLM_EXTRACT ($classified, target="critical_bugs_only") > $critical
S4: ?IF (NOT_EMPTY($critical)) THEN @EXEC_CODE (code="slack_post('#engineering', $critical)", language="python")
S5: @LLM_GENERATE (context=$classified, format="product_management_report") > $report
S6: @WRITE_FILE (path="ticket_report.md", content=$report)
S7: @RESPOND ("Report saved. Alerts sent if critical bugs found.", $report)
PLAN_END

Level 4: Memory-First Patterns

“How tall is Tom Holland?” (query that learns)

PLAN_START
S1: @QUERY_MEMORY (query="Tom Holland height") > $cached
S2: ?IF (NOT_EMPTY($cached)) THEN @RESPOND ($cached)
S3: @WEB_SEARCH (query="Tom Holland height") > $results
S4: @LLM_EXTRACT ($results, target="height") > $height
S5: @STORE_FACT (content="Tom Holland height: $height", importance=0.6)
S6: @RESPOND ("Tom Holland is", $height)
PLAN_END

First time: skips S2 (empty memory), searches, stores, responds. Second time: S2 returns cached answer immediately. Zero web requests.

Level 5: Enterprise Pipeline (FOREACH, ON_FAIL, dot notation, type casting)

“Pull open tickets, classify them, alert on critical bugs, generate report”

PLAN_START
S1: @FETCH_PAGE (url="https://api.hubspot.com/tickets?status=open") > $response:json ON_FAIL TERMINATE ("Ticket API offline")
S2: ?FOREACH ($ticket IN $response.data) THEN @LLM_CLASSIFY ($ticket, categories=["bug", "feature", "critical_bug"]) > $classified
S3: @LLM_EXTRACT ($classified, target="items where category is critical_bug") > $critical
S4: ?IF (NOT_EMPTY($critical)) THEN @EXEC_CODE (code="slack_post('#engineering', $critical)", language="python") ON_FAIL @RETRY(2)
S5: @LLM_GENERATE (context=$classified, format="product_management_report") > $report
S6: @WRITE_FILE (path="ticket_report.md", content=$report)
S7: @RESPOND ("Report saved.", $report.summary)
PLAN_END

Features used:

• > $response:json — type casting ensures the API response is parsed as JSON
• $response.data — dot notation to access the tickets array without an LLM call
• ?FOREACH — classifies each ticket individually
• ON_FAIL TERMINATE — stops cleanly if the API is down
• ON_FAIL @RETRY(2) — retries Slack notification if it fails
• $report.summary — extracts summary from generated report without extra step

“Monitor 3 competitors, compare, save report”

PLAN_START
S1: @WEB_SEARCH (query="competitor A pricing 2026") > $a
S2: @WEB_SEARCH (query="competitor B pricing 2026") > $b
S3: @WEB_SEARCH (query="competitor C pricing 2026") > $c
S4: @LLM_ANALYZE ($a, $b, $c, task="compare pricing strategies") > $analysis
S5: @WRITE_FILE (path="competitor_report.md", content=$analysis)
S6: @STORE_FACT (content="Competitor analysis Q2 2026: $analysis", importance=0.8)
S7: @RESPOND ($analysis)
PLAN_END

Architecture

Comparison

Integration with MCP Suite

LTP is designed to work with the MCP AI Suite:

ltpmcp’s only hard runtime dependencies are pydantic, structlog, click and mcp — the table above shows where the tool_executor / llm_fn callbacks you provide typically delegate in the wider MCP AI Suite. None of those delegated libraries (kernelmcp, websearchmcp, …) is a hard dependency of ltpmcp.

File Structure

ltpmcp/
├── __init__.py       # Exports: LTPPlan, LTPStep, LTPCondition, LTPParser, LTPRuntime,
│                     #          LTPCompiler, validate_plan, plan_to_mermaid,
│                     #          PlanPolicy, PlanVerdict, Violation, verify_plan
├── models.py         # LTPPlan, LTPStep, LTPCondition (with evaluate())
├── parser.py         # LTPParser.parse(raw_text) → LTPPlan | None
├── compiler.py       # LTPCompiler — 1 LLM call, Micro-CLI → LTP
├── runtime.py        # LTPRuntime.execute(plan, tool_executor, llm_fn, ...) → dict
├── analyzer.py       # validate() (exported as validate_plan)
├── visualize.py      # plan_to_mermaid()
├── verifier.py       # verify_plan(plan, PlanPolicy) → PlanVerdict (static, never executes)
├── llm_ops.py        # @LLM_*, @AGENT_CRITIQUE, @CONSENSUS, @SELF_REFINE prompts
├── cli.py            # `ltpmcp` CLI — parse / validate / visualize
├── mcp_server.py     # `ltpmcp-server` — MCP tools: parse_plan, validate_plan, visualize_plan
└── api/              # `ltpmcp-api` — optional FastAPI app ([api] extra)

The package is top-level ltpmcp/. Console-scripts: ltpmcp (CLI), ltpmcp-server (MCP), ltpmcp-api (FastAPI).

Design Principles

1. Compile once, execute deterministically. The LLM’s job is to write the plan, not to run it.

2. The LLM is a tool, not the pilot. @LLM_EXTRACT and @LLM_CLASSIFY are tools in the plan, just like @WEB_SEARCH.

3. Data flows through variables, not context. $results is explicit. No “based on the previous step’s output.”

4. Conditions are code, not prompts. ?IF ($rate > "5") is Python math, not “if the error rate seems high.”

5. Simple tasks stay simple. “hi” = 1 step. “what date” = 2 steps. Don’t over-plan.

6. Complex tasks compose. Each step is atomic and testable. 8 simple steps > 1 complex prompt.

7. Fail gracefully. ON_FAIL provides step-level resilience. @RETRY(N) for transient failures, GOTO for alternative paths, TERMINATE for clean exits.

8. Adapt when needed. @AGENT_SOLVE bridges to ReAct for problems that need exploration. The plan provides structure, the agent provides flexibility.

9. Iterate, don’t loop. ?FOREACH processes lists deterministically. No LLM-driven loops, no token waste on batch operations.

10. Access data directly. $weather.current.temp is free. @LLM_EXTRACT($weather, target="temp") costs an LLM call. Use dot notation first, LLM ops second.

11. Types prevent cascades. > $count:int catches type mismatches early. A string “42” becomes integer 42 before downstream steps try to compare it.

12. Pivot, don’t crash. RE-PLAN lets the plan self-correct when reality diverges from the initial assumptions. 1 recompile call is cheaper than abandoning the plan or falling back to full ReAct.