Agent Server

Drop-in FastAPI wrapper — POST /invoke, POST /stream, GET/DELETE /threads/{id}, GET /health.

For the concepts, start with Agent Server.

AgentServer

AgentServer(agent: Any, title: str = 'Tulip Agent Server', description: str = 'HTTP API for a Tulip AI Agent', api_key: str | None = None, allow_unauthenticated: bool = False, *, stream_tokens: bool = True, metadata_resolver: MetadataResolver | None = None, decision_handler: DecisionHandler | None = None)

Wrap a Tulip Agent as a FastAPI application.

Example

from tulip.agent import Agent, AgentConfig from tulip.server import AgentServer

agent = Agent(config=AgentConfig(system_prompt="Hello", model=model)) server = AgentServer(agent=agent, api_key="secret") server.run(host="127.0.0.1", port=8000)

Wrap agent.

Parameters:

Name	Type	Description	Default
stream_tokens	bool	Stream model token deltas on /stream and /resume (as {"type": "model_chunk", ...} events) when the agent supports it.	True
metadata_resolver	MetadataResolver | None	Server-side source of trusted run metadata, called per request with (request, principal). Its keys override the client's metadata so a client cannot spoof them. Without it, client metadata is passed through as before.	None
decision_handler	DecisionHandler | None	Records the decision payload of a /resume request (see :data:DecisionHandler). Without it, /resume rejects requests that carry a decision.	None

Source code in .sdk/src/tulip/server/app.py

def __init__(
    self,
    agent: Any,
    title: str = "Tulip Agent Server",
    description: str = "HTTP API for a Tulip AI Agent",
    api_key: str | None = None,
    allow_unauthenticated: bool = False,
    *,
    stream_tokens: bool = True,
    metadata_resolver: MetadataResolver | None = None,
    decision_handler: DecisionHandler | None = None,
) -> None:
    """Wrap ``agent``.

    Args:
        stream_tokens: Stream model token deltas on ``/stream`` and
            ``/resume`` (as ``{"type": "model_chunk", ...}`` events) when
            the agent supports it.
        metadata_resolver: Server-side source of trusted run metadata,
            called per request with ``(request, principal)``. Its keys
            override the client's ``metadata`` so a client cannot spoof
            them. Without it, client metadata is passed through as before.
        decision_handler: Records the ``decision`` payload of a
            ``/resume`` request (see :data:`DecisionHandler`). Without it,
            ``/resume`` rejects requests that carry a decision.
    """
    self.agent = agent
    self._stream_tokens = stream_tokens
    self._metadata_resolver = metadata_resolver
    self._decision_handler = decision_handler
    self._title = title
    self._description = description
    # Prefer the explicit arg; fall back to the environment so that
    # deployments don't have to thread the secret through code.
    self._api_key = api_key or os.environ.get("TULIP_SERVER_API_KEY") or None
    self._allow_unauthenticated = allow_unauthenticated
    self._app = None

app property

app: Any

Get or create the FastAPI application.

run

run(host: str = '127.0.0.1', port: int = 8000, **kwargs: Any) -> None

Run the server with uvicorn.

Parameters:

Name	Type	Description	Default
host	str	Bind address. Defaults to loopback — using a non-loopback host requires either api_key to be set or allow_unauthenticated=True on this server.	'127.0.0.1'
port	int	Bind port.	8000
**kwargs	Any	Additional uvicorn.run() arguments.	{}

Source code in .sdk/src/tulip/server/app.py

def run(
    self,
    host: str = "127.0.0.1",
    port: int = 8000,
    **kwargs: Any,
) -> None:
    """Run the server with uvicorn.

    Args:
        host: Bind address. Defaults to loopback — using a
            non-loopback host requires either ``api_key`` to be set
            or ``allow_unauthenticated=True`` on this server.
        port: Bind port.
        **kwargs: Additional uvicorn.run() arguments.
    """
    if self._api_key is None and not self._allow_unauthenticated and not _is_loopback(host):
        msg = (
            f"Refusing to bind AgentServer to {host!r} without an API "
            "key. Set TULIP_SERVER_API_KEY, pass api_key=... to "
            "AgentServer, or pass allow_unauthenticated=True if an "
            "upstream proxy terminates auth."
        )
        raise RuntimeError(msg)

    try:
        import uvicorn
    except ImportError as e:
        msg = "uvicorn is required for AgentServer.run(). Install with: pip install uvicorn"
        raise ImportError(msg) from e

    uvicorn.run(self.app, host=host, port=port, **kwargs)

Serving a graph

GraphRunnable adapts a compiled StateGraph to the runnable interface AgentServer and A2AServer expect. The value at output_key in the graph's final state becomes the reply; with no output_key, the whole final state is stringified. The server's thread_id and metadata are accepted but not yet passed to the graph, and the GET/DELETE /threads/{id} routes don't work for a served graph: they read the wrapped agent's config.checkpointer, which GraphRunnable doesn't have.

GraphRunnable

GraphRunnable(graph: StateGraph | Any, *, input_key: str = 'prompt', output_key: str | None = None, name: str | None = None)

Wrap a graph so :class:AgentServer / :class:A2AServer can publish it.

The adapter speaks the Agent contract — async def run(prompt, *, thread_id=None, metadata=None) -> AsyncIterator[TulipEvent] — by driving graph.stream({input_key: prompt}) underneath. Each per-node StreamEvent becomes a :class:ThinkEvent so SSE / A2A clients see activity in real time; the terminal final-state event becomes a :class:TerminateEvent with final_message set to final_state[output_key] (or the full state stringified when output_key is unset).

Parameters:

Name	Type	Description	Default
graph	StateGraph | Any	A compiled :class:StateGraph (or any object with an async stream(inputs) -> AsyncIterator). Duck-typed so future runnable shapes (compiled pipelines, custom DAGs) slot in unchanged.	required
input_key	str	Key under which the user prompt is placed in the graph's initial state. Defaults to "prompt" because both KEY_PROMPT (deep research) and most StateGraph examples use that name.	'prompt'
output_key	str | None	Key in the graph's final state that holds the user-visible reply. None (default) means stringify the whole final-state dict — usable for diagnostics but most callers will set this explicitly.	None
name	str | None	Optional display name (echoed into ThinkEvent reasoning lines so SSE consumers can tell graphs apart from agents).	None

Notes

• thread_id and metadata from the server are accepted but not propagated yet — graphs don't have a native notion of a server-managed thread. Per-graph checkpointers handle the same concern at a different layer.
• The adapter does NOT touch the agent loop; agents inside the graph keep emitting their own TulipEvent stream onto the tulip event bus. The shim only translates the graph's outer stream.

Source code in .sdk/src/tulip/server/adapters.py

def __init__(
    self,
    graph: StateGraph | Any,
    *,
    input_key: str = "prompt",
    output_key: str | None = None,
    name: str | None = None,
) -> None:
    self.graph = graph
    self.input_key = input_key
    self.output_key = output_key
    self.name = name or getattr(graph, "name", None) or type(graph).__name__

run async

run(prompt: str, *, thread_id: str | None = None, metadata: dict[str, Any] | None = None) -> AsyncIterator[TulipEvent]

Drive self.graph.stream(...) and yield TulipEvent shapes.

Yields:

Name	Type	Description
One	AsyncIterator[TulipEvent]	class:ThinkEvent per intermediate StreamEvent (so
	AsyncIterator[TulipEvent]	consumers see node-by-node progress on the SSE wire), then a
terminal	AsyncIterator[TulipEvent]	class:TerminateEvent with the user-visible reply
	AsyncIterator[TulipEvent]	extracted from the graph's final state.

Source code in .sdk/src/tulip/server/adapters.py

async def run(
    self,
    prompt: str,
    *,
    thread_id: str | None = None,  # noqa: ARG002 — accepted for the Agent contract
    metadata: dict[str, Any] | None = None,  # noqa: ARG002 — same
) -> AsyncIterator[TulipEvent]:
    """Drive ``self.graph.stream(...)`` and yield TulipEvent shapes.

    Yields:
        One :class:`ThinkEvent` per intermediate ``StreamEvent`` (so
        consumers see node-by-node progress on the SSE wire), then a
        terminal :class:`TerminateEvent` with the user-visible reply
        extracted from the graph's final state.
    """
    inputs = {self.input_key: prompt}

    iteration = 0
    last_data: Any = None
    async for ev in self.graph.stream(inputs):
        # ``StreamEvent`` is the graph's per-node envelope. We treat
        # the *last* event as the terminal final-state and only emit
        # a TerminateEvent for it; everything in between becomes a
        # ThinkEvent so the SSE stream stays alive and informative.
        iteration += 1
        last_data = ev.data
        node_id = getattr(ev, "node_id", None)
        mode = getattr(ev, "mode", None)
        label = f"[{self.name}]"
        if node_id:
            label += f" node={node_id}"
        if mode is not None:
            label += f" mode={mode}"
        yield ThinkEvent(
            iteration=iteration,
            reasoning=f"{label} produced data of type {type(ev.data).__name__}",
            tool_calls=[],
        )

    final_message = self._extract_reply(last_data)
    yield TerminateEvent(
        reason="complete",
        iterations_used=iteration,
        final_confidence=1.0,
        total_tool_calls=0,
        final_message=final_message,
    )