Skip to main content

Testing

Internal developers only

The Capture MCP SDK is currently only available to Capture's internal developers. We are working hard on the cybersecurity part, so that we can also offer it to everyone who builds on top of Capture.

The MCP server and the Capture API run in Capture's own cloud, not on your machine. The testing package of each SDK stands in for both in process. A test can start your real service, inspect exactly what it registered and call a tool as a caller you chose, with no MCP server, no Docker and no network beyond loopback.

The two harnesses are the same, member for member in the casing of each language, so a team that uses both learns one API. They stand in for:

  • the registration service of the MCP server, over gRPC
  • the routes of the Capture API that the SDK calls: POST /connect/token, GET /api/me, GET /api/apps/me with /roles, and POST /api/apps/me/permissions
  • the embeddings routes of Captain (see Embeddings)

Your host runs for real against them: registration, finding the app key, declaring permissions, the heartbeat and the permission check in front of your tools.

capture_mcp_sdk.testing adds no dependency: it is the standard library plus the grpcio and httpx that the package already requires.

Never use the testing package from a production path

It contains a fake token endpoint and a fake /api/me. A fake issuer that a deployed process can reach is a security problem, not a convenience.

The harness also plays the calling side, which is the part you cannot otherwise reach. call_tool (Python) and CallToolAsync (.NET) dial the callback address that your service registered and forward a bearer token you supply, exactly as the MCP server does. The mock passes your token through untouched, so it stands in for the delegated token. It does not model the narrowing to what the person and your app may both do: the Capture API applies that.


Test a tool​

import pytest
from capture_mcp_sdk import CaptureMcpCallContext, add_capture_mcp_tools
from capture_mcp_sdk.testing import MockCaptureMcpServer, mock_settings


async def test_my_tool_is_permission_gated():
async with MockCaptureMcpServer() as mock:
app = add_capture_mcp_tools(mock_settings(mock, service_name="sandbox"))

@app.tool(requires_permission="events:view", read_only=True)
async def list_events(company_id: int, ctx: CaptureMcpCallContext) -> dict:
"""Lists events for a company."""
return {"company_id": company_id, "is_owner": ctx.is_owner(company_id)}

allowed = mock.create_caller().grant(42, "events:view")
denied = mock.create_caller().grant(42, "events:write")

await app.start()
try:
registration = await mock.wait_for_registration()
assert registration.tool("sandbox_list_events").required_permissions == (f"{mock.app_key}:events:view",)

ok = await mock.call_tool("sandbox_list_events", {"companyId": 42}, allowed)
assert ok.result == {"company_id": 42, "is_owner": False}

nope = await mock.call_tool("sandbox_list_events", {"companyId": 42}, denied)
assert nope.permission_denied and nope.company_id == 42
finally:
await app.stop()

mock_settings(mock, ...) fills api_base_url and gateway_address from the mock and picks a free callback_address port. Build a CaptureMcpSettings by hand if you prefer.


What the mock exposes​

Python.NETWhat it is for
gateway_address, api_base_urlGatewayAddress, ApiBaseUrlWhat to point your settings at.
app_key (get and set)AppKey (get and set)The prefix the SDK adds to declared permissions. None or null means no role declared yet: set it halfway through a test and the deferred registration of the SDK recovers on its next retry.
api_available (get and set)ApiAvailable (get and set)With False, every route answers 503, which stands in for an outage of the Capture API. Your host must still start.
create_caller(token=None)CreateCaller(token?)A caller that /api/me recognises. Chain .grant(company_id, "events:view") (.Grant in .NET), .grant_namespaced(...) and .owns(company_id).
registrations, unregistered_services, declared_permissionsRegistrations, UnregisteredServices, DeclaredPermissionsEverything that was received so far.
wait_for_registration(count=1)WaitForRegistrationAsync(count, timeout)Waits for the nth registration. Use count 2 to see the heartbeat.
wait_for_declared_permissions(count), wait_for_unregister(service)WaitForDeclaredPermissionsAsync(count), WaitForUnregisterAsync(serviceName)The same, for the permission catalog and for shutdown.
call_tool(tool_name, arguments, caller)CallToolAsync(tool, args, caller)Plays the MCP server. It calls your tool through the gRPC callback.
declaration_attempts, me_call_countPython only: the duplicates that the SDK retried, and whether the cache for callers held.

A recorded registration is decoded into plain values, not protobuf messages, so you assert on what the MCP server would have stored:

  • A registration (MockToolRegistration) has the service name, the callback address, the TTL, the bearer token of the registration, and its tools, with Tool(name) to find one.
  • A tool (MockRegisteredTool) has the name, description, input and output schema, the required permissions, whether it has annotations, and read_only, destructive, idempotent and open_world.
  • A call (MockToolCallResult) has the result as JSON, whether it is an error, the error message, whether the permission was denied, the required permissions and the company id.

Two behaviours that are mocked the way the real services behave​

  • grant takes bare keys, and the mock adds the app key, exactly as the Capture API does. A test never hard-codes the prefix that Capture assigns, and the same test still passes if the app key changes. Use grant_namespaced to check against the permission of another app.
  • Caller tokens are unsigned and shaped like a JWT (alg: none, with a real iss, sub and exp). The SDK reads iss and exp from the unverified payload to scope and cap its cache for callers, so an opaque token would leave that behaviour untested. They are not credentials and validate nowhere.

Python adds a third: a permission that is declared again answers 400 APP_DEFINITION_INVALID, like the real endpoint. The SDK has to treat that as success, and answering 200 would hide a regression in exactly that handling.


What to test​

A test with the mock can cover:

  • the registration payload, with namespaced permissions and the injected companyId
  • the machine token on the registration channel, and the declaration of the permission catalog
  • an allowed call, every reason for a denial and a call without authentication
  • the heartbeat, and a graceful unregister
  • a deferred registration that recovers once the app key is set
  • a start that survives an outage of the Capture API

Run the mock as a local process​

To point a running app, yours or an MCP client, at something instead of writing a test, bind fixed ports and leave the mock up. Binding to a wildcard address such as 0.0.0.0 is not dialable, so say which address to advertise.

# mock_gateway.py: python mock_gateway.py
import asyncio
from capture_mcp_sdk.testing import MockCaptureMcpServer


async def main() -> None:
async with MockCaptureMcpServer(gateway_port=6001, api_port=5001, app_key="app-local") as mock:
mock.create_caller("dev-token").grant(1, "events:view").owns(1)
print(f"gateway_address={mock.gateway_address}")
print(f"api_base_url={mock.api_base_url}")
await asyncio.Event().wait() # serve until Ctrl-C


asyncio.run(main())

Configure your app with the two printed values. Any app_client_id and app_client_secret will do, because the fake /connect/token issues a token to anyone. Call a tool with authorization: Bearer dev-token.

To let another machine or container reach the mock, bind every interface and say which address to advertise:

MockCaptureMcpServer(host="0.0.0.0", advertise_host="192.168.1.10", ...)

The standalone mock does not speak MCP to an agent. It is the registration and callback side only. To drive tools from an AI tool, connect to the real MCP server, as described under MCP.