agentstatus-mcpjobgraph

agentstatus-mcpjobgraph

Stop paying for out-of-order MCP tool calls.

Agents ignore "call X first" in your tool descriptions. They still hit get_package / fetch / get_deal with garbage IDs. You still eat the traffic, the error noise, and the bad demos.

agentstatus-mcpjobgraph is a thin author-side gate: illegal order never reaches your tools. Legal order still runs, and binds can fill downstream args from earlier steps.

Measured on public MCPs (2026-09-01): 7/7 eligible servers — bare MCP got the illegal call; middleware blocked it with upstream never hit. See results/friction_top9_2026-09-01.md.

Install

pip install agentstatus-mcpjobgraph
# optional adapters
pip install "agentstatus-mcpjobgraph[fastmcp]"
pip install "agentstatus-mcpjobgraph[mcp]"

Python 3.10+. Core has zero required dependencies.

From AgentStatus / Carmel Labs (agentstatus.dev).

60-second FastMCP drop-in

Hand-author the job. Do not ship inferred graphs without reading them.

from fastmcp import FastMCP
from agentstatus_mcpjobgraph import Workflow, install

mcp = FastMCP("packages")

@mcp.tool
def search_packages(query: str) -> dict:
    return {"package_id": "abc123", "name": query}

@mcp.tool
def get_package(package_id: str) -> dict:
    return {"package_id": package_id, "ok": True}

install(
    mcp,
    [
        Workflow(
            id="search_then_get",
            steps=["search_packages", "get_package"],
            binds={"package_id": "search_packages.package_id"},
        )
    ],
)

if __name__ == "__main__":
    mcp.run()

Calling get_package first → blocked with a repair hint. Never hits your function.

Official MCP Python SDK

from mcp.server import Server
from agentstatus_mcpjobgraph import Workflow, make_enforcer, jobgraph_middleware

server = Server("packages")
enforcer = make_enforcer([
    Workflow(id="search_then_get", steps=["search_packages", "get_package"],
             binds={"package_id": "search_packages.package_id"}),
])
server.middleware.append(jobgraph_middleware(enforcer))

Or wrap a classic handler:

from agentstatus_mcpjobgraph import make_enforcer, Workflow, wrap_call_tool

enforcer = make_enforcer([Workflow(id="x", steps=["a", "b"])])

async def handle(name: str, arguments: dict):
    ...

handle = wrap_call_tool(enforcer, handle)

In-process decorator (@with_jobs)

from agentstatus_mcpjobgraph import Workflow, make_enforcer, with_jobs

enforcer = make_enforcer([Workflow(id="x", steps=["a", "b"])])

@with_jobs(enforcer, session_id="default")
def call_tool(name: str, arguments: dict):
    return dispatch(name, arguments)

Hand-authored workflows (preferred)

from agentstatus_mcpjobgraph import Workflow, load_workflows

workflows = [
    Workflow(
        id="search_then_get",
        steps=["search_packages", "get_package"],
        binds={"package_id": "search_packages.package_id"},
    )
]
# or
workflows = load_workflows("workflows.json")

workflows.json:

{
  "workflows": [
    {
      "id": "search_then_get",
      "steps": ["search_packages", "get_package"],
      "binds": {"package_id": "search_packages.package_id"}
    }
  ]
}

Bootstrap only (then edit)

Infer candidates from a live catalog, then rewrite by hand:

from agentstatus_mcpjobgraph import bootstrap_workflows_from_tools

sketches = bootstrap_workflows_from_tools(tools, instructions)
# review / edit sketches → ship as Workflow(...)

Inference is a sketch. Enforcement should run on jobs you own.

HTTP proxy (no code change on the origin)

python -m agentstatus_mcpjobgraph proxy --url https://your-mcp.example/mcp --port 8765

Point clients at http://127.0.0.1:8765/mcp.

Prove friction removal

python -m agentstatus_mcpjobgraph friction --url https://chat.gitsim.com/api/mcp

Expect friction_removed=True: illegal order hits upstream without the gate; blocked with upstream_hit=false with the gate.

What this is / is not

Emit-only _meta.workflows without a gate does not remove friction. Hosts ignore it. This package is the gate.

Package layout

agentstatus_mcpjobgraph/
  workflow.py       # hand-authored Workflow (preferred)
  middleware.py     # JobEnforcer core
  with_jobs.py      # @with_jobs + install()
  fastmcp_ext.py    # FastMCP middleware
  mcp_sdk.py        # official MCP Python SDK middleware
  bootstrap.py      # infer once, then edit (not the default)

License

MIT