Adapters

External-framework frontends for APAB's MCP tools. Each has its own optional extra (apab[strands], apab[langgraph]).

apab.adapters.strands

Strands Agents adapter: use APAB's MCP tools from a Strands agent.

The adapter talks to APAB over its public MCP surface: it launches apab mcp serve --transport stdio as a subprocess and hands the connection to Strands' MCPClient. This keeps the integration on the stable MCP protocol rather than APAB's in-process tool dispatcher, whose FastMCP internals are private.

Requires the strands extra::

pip install "apab[strands]"

Typical use::

from strands import Agent
from apab.adapters.strands import apab_mcp_client, apab_system_prompt

client = apab_mcp_client(config_path="apab.yaml")
with client:
    agent = Agent(
        model=...,  # any Strands model provider
        tools=client.list_tools_sync(),
        system_prompt=apab_system_prompt(),
    )
    agent("Design a 28 GHz 8x8 patch array and report its metrics.")

apab_server_parameters

apab_server_parameters(config_path=None, env=None)

Build MCP StdioServerParameters that launch APAB's server.

The server runs in a subprocess with the same Python interpreter, so it sees the same installed apab and its tools. Observability env vars (TRACEPARENT, APAB_OBSERVABILITY, APAB_TRACE_JSONL) are forwarded automatically; env entries override them. The MCP client merges these on top of its safe default environment.

Source code in src/apab/adapters/strands.py

def apab_server_parameters(
    config_path: str | Path | None = None,
    env: dict[str, str] | None = None,
) -> Any:
    """Build MCP ``StdioServerParameters`` that launch APAB's server.

    The server runs in a subprocess with the same Python interpreter,
    so it sees the same installed apab and its tools. Observability env
    vars (``TRACEPARENT``, ``APAB_OBSERVABILITY``, ``APAB_TRACE_JSONL``)
    are forwarded automatically; *env* entries override them. The MCP
    client merges these on top of its safe default environment.
    """
    from mcp import StdioServerParameters

    args = ["-m", "apab.cli", "mcp", "serve", "--transport", "stdio"]
    if config_path is not None:
        args += ["--config", str(config_path)]
    merged = _observability_env()
    if env:
        merged.update(env)
    return StdioServerParameters(
        command=sys.executable, args=args, env=merged or None
    )

apab_mcp_client

apab_mcp_client(config_path=None, env=None)

Return a Strands MCPClient connected to APAB over stdio.

Use it as a context manager; tools are available inside the block via client.list_tools_sync().

Source code in src/apab/adapters/strands.py

def apab_mcp_client(
    config_path: str | Path | None = None,
    env: dict[str, str] | None = None,
) -> MCPClient:
    """Return a Strands ``MCPClient`` connected to APAB over stdio.

    Use it as a context manager; tools are available inside the block
    via ``client.list_tools_sync()``.
    """
    _require_strands()
    from mcp.client.stdio import stdio_client
    from strands.tools.mcp import MCPClient

    params = apab_server_parameters(config_path, env=env)
    return MCPClient(lambda: stdio_client(params))

apab_system_prompt

apab_system_prompt(config=None)

APAB's own agent system prompt, reusable for a Strands agent.

Source code in src/apab/adapters/strands.py

def apab_system_prompt(config: dict[str, Any] | None = None) -> str:
    """APAB's own agent system prompt, reusable for a Strands agent."""
    from apab.agent.prompts import build_system_prompt

    return build_system_prompt(config)

apab.adapters.langgraph_pipeline

Deterministic LangGraph pipeline over APAB's engineering tools.

Where the agent orchestrator lets an LLM pick tools turn by turn, this pipeline runs a fixed engineering sequence with explicit state, checkpointing, and streaming progress:

validate_config
-> unit_cell (only with EdgeFEM installed and configured)
-> pattern
-> system_eval
-> constraint_check
-> plots
-> report

Nodes are plain Python callables that dispatch APAB's MCP tools in-process; no LLM is involved. Results land in a normal APAB run bundle (manifest.json, artifacts/), and each node is wrapped in an apab.node.<name> span when observability is enabled.

Requires the langgraph extra::

pip install "apab[langgraph]"

Note: langgraph depends on langchain-core; APAB uses no LangChain model wrappers.

PipelineState

Bases: TypedDict

State threaded through the pipeline graph.

Source code in src/apab/adapters/langgraph_pipeline.py

class PipelineState(TypedDict, total=False):
    """State threaded through the pipeline graph."""

    config: dict[str, Any]
    run_id: str
    run_dir: str
    unit_cell: dict[str, Any]
    pattern: dict[str, Any]
    system: dict[str, Any]
    violations: list[str]
    plots: list[str]
    report_path: str
    errors: list[str]

Scenario dataclass

System-evaluation scenario parameters for the system_eval node.

Source code in src/apab/adapters/langgraph_pipeline.py

@dataclass
class Scenario:
    """System-evaluation scenario parameters for the system_eval node."""

    bandwidth_hz: float = 100e6
    range_m: float = 1000.0
    tx_power_w_per_elem: float = 1.0
    scenario_type: str = "comms"
    required_snr_db: float = 10.0

Constraints dataclass

Metric thresholds checked by the constraint_check node.

Source code in src/apab/adapters/langgraph_pipeline.py

@dataclass
class Constraints:
    """Metric thresholds checked by the constraint_check node."""

    min_directivity_dbi: float | None = None
    max_sidelobe_level_db: float | None = None
    extra: dict[str, float] = field(default_factory=dict)

build_pipeline

build_pipeline(config, *, scenario=None, constraints=None, workspace=None, checkpoint=True)

Compile the pipeline graph.

Returns (graph, run_ctx, initial_state). Invoke with::

graph.invoke(initial_state, config={"configurable": {"thread_id": run_ctx.run_id}})

or stream node-by-node with graph.stream(..., stream_mode="updates"). With checkpoint true, state persists to <run_dir>/checkpoint.sqlite keyed by thread_id, so a rerun with the same thread id resumes rather than recomputes.

Source code in src/apab/adapters/langgraph_pipeline.py

def build_pipeline(
    config: ProjectConfig | dict[str, Any],
    *,
    scenario: Scenario | None = None,
    constraints: Constraints | None = None,
    workspace: Workspace | None = None,
    checkpoint: bool = True,
) -> tuple[Any, RunContext, PipelineState]:
    """Compile the pipeline graph.

    Returns ``(graph, run_ctx, initial_state)``. Invoke with::

        graph.invoke(initial_state, config={"configurable": {"thread_id": run_ctx.run_id}})

    or stream node-by-node with ``graph.stream(..., stream_mode="updates")``.
    With ``checkpoint`` true, state persists to
    ``<run_dir>/checkpoint.sqlite`` keyed by ``thread_id``, so a rerun
    with the same thread id resumes rather than recomputes.
    """
    try:
        from langgraph.graph import END, START, StateGraph
    except ImportError as exc:
        raise ImportError(
            "The LangGraph pipeline requires the langgraph package. "
            "Install it with: pip install 'apab[langgraph]'"
        ) from exc

    if isinstance(config, dict):
        config = ProjectConfig.model_validate(config)

    workspace = workspace or Workspace(Path(config.project.workspace))
    workspace.ensure_dirs()
    run_ctx = workspace.new_run()

    mode = config.llm.redaction_mode
    dispatcher = ToolDispatcher(
        redaction_mode=mode.value if hasattr(mode, "value") else str(mode),
    )
    nodes = _Nodes(
        dispatcher,
        run_ctx,
        scenario or Scenario(),
        constraints or Constraints(),
    )

    graph = StateGraph(PipelineState)
    graph.add_node("validate_config", nodes.validate_config)
    graph.add_node("unit_cell", nodes.unit_cell)
    graph.add_node("pattern", nodes.pattern)
    graph.add_node("system_eval", nodes.system_eval)
    graph.add_node("constraint_check", nodes.constraint_check)
    graph.add_node("plots", nodes.plots)
    graph.add_node("report", nodes.report)

    def route_after_validate(state: PipelineState) -> str:
        if state.get("errors"):
            return "report"
        cfg = state["config"]
        wants_edgefem = (
            (cfg.get("solver") or {}).get("backend") == "edgefem"
            and cfg.get("unit_cell") is not None
        )
        if wants_edgefem and _edgefem_available():
            return "unit_cell"
        return "pattern"

    graph.add_edge(START, "validate_config")
    graph.add_conditional_edges(
        "validate_config",
        route_after_validate,
        {"unit_cell": "unit_cell", "pattern": "pattern", "report": "report"},
    )
    graph.add_edge("unit_cell", "pattern")
    graph.add_edge("pattern", "system_eval")
    graph.add_edge("system_eval", "constraint_check")
    graph.add_edge("constraint_check", "plots")
    graph.add_edge("plots", "report")
    graph.add_edge("report", END)

    checkpointer = None
    if checkpoint:
        import sqlite3

        from langgraph.checkpoint.sqlite import SqliteSaver

        conn = sqlite3.connect(
            run_ctx.run_dir / "checkpoint.sqlite", check_same_thread=False,
        )
        checkpointer = SqliteSaver(conn)

    compiled = graph.compile(checkpointer=checkpointer)

    initial_state: PipelineState = {
        "config": config.model_dump(mode="json"),
        "run_id": run_ctx.run_id,
        "run_dir": str(run_ctx.run_dir),
        "errors": [],
    }
    return compiled, run_ctx, initial_state

run_pipeline

run_pipeline(config, *, scenario=None, constraints=None, workspace=None, checkpoint=True)

Build and run the pipeline; returns the final state.

Source code in src/apab/adapters/langgraph_pipeline.py

def run_pipeline(
    config: ProjectConfig | dict[str, Any],
    *,
    scenario: Scenario | None = None,
    constraints: Constraints | None = None,
    workspace: Workspace | None = None,
    checkpoint: bool = True,
) -> PipelineState:
    """Build and run the pipeline; returns the final state."""
    from apab.observability import init_observability, shutdown_observability

    graph, run_ctx, initial_state = build_pipeline(
        config,
        scenario=scenario,
        constraints=constraints,
        workspace=workspace,
        checkpoint=checkpoint,
    )

    cfg = config if isinstance(config, ProjectConfig) else ProjectConfig.model_validate(config)
    init_observability(cfg.observability, run_ctx=run_ctx)
    try:
        with span("apab.pipeline", **{"apab.run_id": run_ctx.run_id}):
            result: PipelineState = graph.invoke(
                initial_state,
                config={"configurable": {"thread_id": run_ctx.run_id}},
            )
        return result
    finally:
        shutdown_observability()