Developing with the SDK

The other SDK pages tell you what the surface is. This one is about the loop you actually work in: write a handler, test it in a second with no cluster, then ship it as an image.

The short version: an agent is a container that answers the launcher’s invoke contract. The SDK is typed sugar over the localhost plane the launcher injects around your process. So the fast inner loop is the one where you replace that plane with fakes and never leave your editor — and the SDK ships those fakes.

Getting the SDK

Inside an agent image, it is already there. base-python and base-node bundle the package, so a FROM ghcr.io/ctxmesh/base-python:v0.1.0-beta.8 agent can import ctxmesh with no install step and no requirements.txt entry.

Your image must run as a non-root user

The controller hardens every agent container with runAsNonRoot: true, so an image whose user is root never starts — the pod reports:

CreateContainerConfigError: container has runAsNonRoot and image will run as root

Fixed in v0.1.0-beta.7. base-python and base-node now end as 65532:65532, so an agent built on them needs nothing extra. If you are pinned to v0.1.0-beta.6 or earlier, those images run as root — either move to v0.1.0-beta.8, set spec.unconfined: true on the AgentDeployment, or add the user yourself:

FROM ghcr.io/ctxmesh/base-python:v0.1.0-beta.6
USER 65532:65532

On v0.1.0-beta.8 you only need to act if your image writes to a root-owned path at build time. Then step up and back down — whatever user the image ends on is the user the container runs as:

FROM ghcr.io/ctxmesh/base-python:<version>
WORKDIR /app
COPY requirements.txt /app/requirements.txt
USER root
RUN pip install --no-cache-dir -r /app/requirements.txt
USER 65532:65532
COPY agent.py /app/agent.py

Use the numeric uid, not a name: the kubelet cannot check a non-numeric image user against runAsNonRoot and refuses the container for that reason instead. If your agent writes to its working directory at runtime, give it ownership — COPY --chown=65532:65532 — or write to /tmp.

On your machine, install it from the repository:

git clone https://github.com/ctxmesh/ctxmesh
pip install ./ctxmesh/sdk/python

# TypeScript
cd ctxmesh/sdk/typescript && pnpm install && pnpm build

All six SDKs are published, and a registry install is the normal path:

pip install ctxmesh                 # Python
npm install ctxmesh@beta            # TypeScript
go get github.com/ctxmesh/ctxmesh/sdk/go
cargo add ctxmesh                   # Rust
gem install ctxmesh --pre           # Ruby — gem install needs --pre while in beta

Java takes the coordinates ai.ctxmesh:ctxmesh from Maven Central.

Build from source only when you are changing the SDK itself. On npm, always use @beta. No stable release exists yet, so latest is pinned to an older pre-release and a bare npm install ctxmesh will hand you a version behind the others.

Test with no cluster

ctxmesh.testing ships fakes of the localhost plane: tiny http.server stubs that speak the same contracts the launcher does. Point a client at them and your handler runs end-to-end — no cluster, no launcher, no provider key, no network.

from ctxmesh import InvokeRequest, agent
from ctxmesh.config import PlaneConfig, RunContext
from ctxmesh.serve import process_invoke
from ctxmesh.testing import DiscoveryStub, GatewayStub, MemoryStub


def handler(req: InvokeRequest) -> str:
    """The agent under test — the same function you'd serve in production."""
    return req.client.model.chat(
        "gpt-4o-mini", [{"role": "user", "content": req.input}]
    ).text


def test_the_agent_answers():
    with (
        MemoryStub() as mem,
        DiscoveryStub() as disc,
        GatewayStub(content="hello from offline") as gw,
    ):
        cfg = PlaneConfig.for_test(
            memory_base_url=mem.base_url,
            discovery_base_url=disc.base_url,
            model_gateway_url=gw.base_url,
            run=RunContext(agent_name="offline-agent", conversation_id=""),
        )
        client = agent.from_config(cfg)

        body = process_invoke(client, handler, "offline-agent", b'{"input":"hi"}', {})
        assert body["output"] == "hello from offline"

Each stub binds an ephemeral localhost port and exposes it as base_url, so tests run in parallel without fighting over ports. PlaneConfig.for_test is the seam: it builds a fully wired config from explicit URLs instead of reading the launcher’s environment.

The four stubs cover the plane:

Stub	Fakes	Useful knobs
`MemoryStub`	session + long-term memory (`:2998`)	— round-trips what you write
`DiscoveryStub`	the tool manifest (`:2999`) and a real inline MCP server	`tool_result`, `manifest_input_schema`
`GatewayStub`	the OpenAI-compatible model gateway	`content`, `usage`, `model`, `force_status`
`FeedbackStub`	feedback scores (`:2995`)	`force_status`

force_status is how you test the paths that are hard to provoke for real — a budget 402 from the gateway, a 502 from a feedback backend. Those branches are usually the least exercised code in an agent and the most annoying to reproduce in a cluster.

Assert what the agent did, not just what it returned

This is the part worth learning. Every stub records each request it received, so a test can check the agent talked to the plane correctly — not merely that the return value looked right. A handler that returns a plausible string while calling the wrong endpoint, dropping the conversation id, or sending the wrong model is exactly the bug an output assertion misses.

def test_the_agent_uses_the_conversation_it_was_given():
    with MemoryStub() as mem, DiscoveryStub() as disc, GatewayStub() as gw:
        cfg = PlaneConfig.for_test(
            memory_base_url=mem.base_url,
            discovery_base_url=disc.base_url,
            model_gateway_url=gw.base_url,
            run=RunContext(agent_name="offline-agent", conversation_id=""),
        )
        client = agent.from_config(cfg).with_conversation("conv-1")

        client.memory.append({"role": "user", "content": "remember me"})
        assert client.memory.get() == [{"role": "user", "content": "remember me"}]

        # The plane was really exercised, on the right conversation.
        wrote = [r for r in mem.requests if r.method == "POST"]
        assert wrote, "the handler never wrote to memory"
        assert wrote[-1].path == "/memory/conv-1/append"
        assert wrote[-1].json() == {"role": "user", "content": "remember me"}

A RecordedRequest carries method, path, query, headers and body, plus .json() for the parsed body. Assert on whichever carries meaning for your agent: the model name you sent, the tool you resolved, the conversation you scoped to. Note that the conversation id travels in the memory path, not a header — which is exactly the kind of thing worth pinning in a test, because it is invisible in the return value. (headers keys are lowercased, as http.server delivers them.)

DiscoveryStub has a deliberate trap built in, and it is a good one. Its catalog advertises the tool as word-count (hyphen) while the MCP server exposes word_count (underscore) — the same split a real deployment has. A client that fails to resolve the catalog key to the real MCP name gets a JSON-RPC error from the stub instead of a silent pass. The fake is built to fail you the way production would.

What the fakes deliberately do not cover

They stand in for the launcher plane, not for the platform. Guardrails, approval gating, quotas, the deploy gate, real tracing backends, cost attribution and tenant isolation are all enforced by the control plane around your pod, and none of it is in scope for a unit test. A green offline suite means your handler speaks the plane correctly; it says nothing about whether the policy in front of it will let a given call through.

That is the right split — but it does mean the offline suite is not the whole story. Deploy to a cluster before you believe it.

From a passing test to a running agent

The handler you tested is the handler you serve:

from ctxmesh import agent, serve

client = agent.from_env()          # reads the launcher-injected environment
serve(client, handler)             # the same `handler` the test drove

Then build and deploy:

docker build -t your-registry/your-agent:v1 .

apiVersion: agents.ctxmesh.ai/v1beta1
kind: AgentDeployment
metadata:
  name: your-agent
spec:
  image: your-registry/your-agent:v1

from_env() and from_config() produce the same client — the only difference is where the plane’s URLs come from. That is what makes the offline test meaningful: production swaps the config source, not the code path.

For iterating against a real launcher and console without a full cluster, see local development. For deploying properly, see deploy an agent.