Test your agents

guardian_agent.testing.FakeProxy stands in for viper proxy in your tests. It serves real HTTP on 127.0.0.1, so your agents, and the OpenAI and Anthropic clients bound to them, run exactly as they do in production, with no Viper, no keys and no network. It answers the agent API and model requests as the proxy does, with the same error codes, and records what your agents sent.

from collections.abc import Iterator

import openai
import pytest
from openai.types.chat import ChatCompletionToolParam

import guardian_agent as ga
from guardian_agent.testing import FakeProxy

INSTRUCTIONS = "You research questions and cite your sources."


@ga.tool
def search_docs(query: str) -> list[str]:
    """Search the internal documentation."""
    return []


# The tool as the definition records it, and as the request must send it
SEARCH = ga.Tool.from_function(search_docs)
TOOLS: list[ChatCompletionToolParam] = [
    {
        "type": "function",
        "function": {
            "name": SEARCH.name,
            "description": SEARCH.description,
            "parameters": SEARCH.input_schema,
        },
    }
]


@pytest.fixture
def proxy() -> Iterator[FakeProxy]:
    with FakeProxy(agents=["researcher"]) as proxy:
        yield proxy


def researcher(proxy: FakeProxy) -> ga.Agent:
    return ga.Guardian(proxy.url).agent(
        "researcher",
        model=ga.Model("openai", "gpt-4.1"),
        instructions=INSTRUCTIONS,
        tools=[search_docs],
    )


def ask(client: openai.OpenAI) -> str | None:
    reply = client.chat.completions.create(
        model="gpt-4.1",
        messages=[
            {"role": "system", "content": INSTRUCTIONS},
            {"role": "user", "content": "Summarise the release notes."},
        ],
        tools=TOOLS,
    )
    return reply.choices[0].message.content


def test_the_researcher_is_named_and_described(proxy: FakeProxy) -> None:
    ask(researcher(proxy).bind(openai.OpenAI(api_key="test")))

    sent = proxy.requests[-1]
    assert sent.agent == "researcher"
    # What the request offered the model is what the definition records
    declared = [tool["name"] for tool in proxy.definitions["researcher"]["tools"]]
    offered = [tool["function"]["name"] for tool in sent.json["tools"]]
    assert offered == declared == ["search_docs"]


def test_a_lost_answer_is_reported(proxy: FakeProxy) -> None:
    client = researcher(proxy).bind(openai.OpenAI(api_key="test"))
    proxy.refuse("upstream_response_lost")

    with pytest.raises(openai.APIStatusError) as raised:
        ask(client)

    assert isinstance(ga.proxy_error(raised.value), ga.UpstreamResponseLost)

Use it as a context manager, as the fixture above does, or call start() and close() yourself. It’s safe to use from many threads and async tasks at once. The example needs pytest, and the model clients your agents use, in your test environment.

What it checks

It checks a registration the way the proxy does:

• the agent must be enrolled, that is, named in agents;
• the definition must pass the schema, the proxy’s other rules and the credential scan;
• a model request that names an agent must name an enrolled one, a definition registered for it, and a sub-agent the definition declares. A request that names no agent passes, unless the proxy is strict.

It takes skill and instructions files as the proxy does (guardian.skill(), guardian.instructions()), with the proxy’s checks, and refuses a definition naming files it doesn’t hold.

What it answers

OpenAI’s Chat Completions and Responses APIs and Anthropic’s Messages API get a short canned reply, streamed if the request asks for a stream. To answer with your own JSON, use proxy.reply(body): it answers every model request from then on, in place of the canned replies, with status= to answer with another HTTP status, such as a provider’s error. body can also be a function that takes the FakeRequest, for replies that change from call to call, such as a tool call and then an answer. Your own replies are never streamed.

What it records

• proxy.requests holds every model request that got through, as a FakeRequest: its method, path, headers, body and parsed json, and the agent, definition_cid and subagent it named. The proxy removes the agent, definition and sub-agent headers before forwarding, so they aren’t in headers.
• proxy.definitions holds each agent’s latest registered definition as JSON, keyed by agent name, with the schema’s field names.

So a test can check what an agent sent, as well as what it declared. With a model client you call yourself, comparing the two is how you check your requests send the instructions and tools the definition records.

Refusals, restarts and older proxies

• Refusals: proxy.refuse("upstream_response_lost") makes the next model request fail with that code, as the proxy would, so you can test how your code handles it. Clients retry signing_failed, internal_error and upstream_unreachable, so pass times= to refuse several requests in a row and outlast their retries, or give the client max_retries=0. They don’t retry the example’s upstream_response_lost, which the SDK marks final, since the model may already have run.
• Restarts: proxy.restart() forgets every registered definition, as a restarted proxy does. The SDK registers again and resends the request.
• Strict mode: FakeProxy(agents=[...], strict=True) refuses model requests that name no agent, as viper proxy --require-agent-header does.
• An older proxy: FakeProxy(features=[...]) says it does only those features, so Guardian refuses it as it would a Viper that lacks the rest. Leave one out to see how your code meets an older proxy.

What it doesn’t do

FakeProxy signs nothing, calls no model, stores no prompts apart, and forwards nothing anywhere. It doesn’t compute real content IDs: each definition gets a stable stand-in (fake-…), and each skill’s or instructions’ files one in a collection CID’s form (bagaachrafake…), the same for the same content. Only the real proxy computes real ones.