Skip to content
JustAPI

Agent System (MCP Tools & Sessions)

JustAPI includes a first-class agent-native system that differentiates it from traditional web frameworks. It provides MCP tools, durable sessions, and validated streaming — all owned by Rust.

MCP Tools (@app.tool)

Section titled “MCP Tools (@app.tool)”

Register Python functions as MCP-compatible tools. Schemas are inferred from type hints:

from justapi import JustAPIApp


app = JustAPIApp()




@app.tool
def add(a: int, b: int) -> int:
    """Add two integers."""
    return a + b




@app.tool
def search_docs(query: str, top_k: int = 3) -> list:
    """Search documentation."""
    return [{"text": "JustAPI runs routing in Rust", "score": 0.95}]

Tool Discovery & Invocation

Section titled “Tool Discovery & Invocation”

Tools are automatically exposed via system routes:

Terminal window
# List all registered tools
GET /_system/tools


# Call a tool
POST /_system/tools/call
Content-Type: application/json


{"name": "add", "arguments": {"a": 5, "b": 7}}

MCP Stdio Server

Section titled “MCP Stdio Server”
app.run_mcp_stdio()  # Run an MCP stdio server for local AI tool use

Durable Sessions (app.enable_sessions())

Section titled “Durable Sessions (app.enable_sessions())”

Rust-backed per-client state storage:

app.enable_sessions()




@app.get("/agent/state")
def agent_state(request, session: Session):
    prior = session.get().get("runs", 0)
    session.update(runs=prior + 1)
    return {"session_id": session.id, "runs": prior + 1}

Sessions are identified by the justapi_session cookie or ?session= query parameter.

Validated Streaming (@app.stream_json)

Section titled “Validated Streaming (@app.stream_json)”

Stream generator output as NDJSON or JSON array, validating every item in Rust:

STEP_SCHEMA = {
    "type": "object",
    "properties": {
        "step": {"type": "string"},
        "result": {},
    },
    "required": ["step", "result"],
}




@app.stream_json("/agent/run", schema=STEP_SCHEMA, mode="ndjson")
def agent_run(request, session: Session):
    yield {"step": "think", "result": "Analyzing..."}
    yield {"step": "act", "result": "Performing task..."}
    yield {"step": "observe", "result": "Task complete"}

Invalid items abort the stream — clients never see partial or invalid data.

Complete Example

Section titled “Complete Example”
from justapi import JustAPIApp, Session


app = JustAPIApp()
app.enable_sessions()
app.enable_system_routes()




@app.tool
def add(a: int, b: int) -> int:
    return a + b




@app.stream_json("/run", schema=STEP_SCHEMA, mode="ndjson")
def run(request, session: Session):
    result = add(2, 3)
    prior = session.get().get("runs", 0)
    session.update(runs=prior + 1)
    yield {"step": "compute", "result": result}




if __name__ == "__main__":
    app.run("127.0.0.1:8000")

See Also

Section titled “See Also”