Testing
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/mewith/roles, andPOST /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.
- Python
- .NET
capture_mcp_sdk.testing adds no dependency: it is the standard library plus the grpcio and httpx that the package already requires.
The package is Capture.Cloud.Mcp.Sdk.Testing. It is separate on purpose, and it never references Capture.Cloud.Mcp.Sdk, so it cannot assert the behaviour of the SDK by construction.
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
- Python
- .NET
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.
await using var mock = await CaptureMcpMockServer.StartAsync();
// your own host, configured against the mock
var builder = WebApplication.CreateBuilder();
builder.WebHost.ConfigureKestrel(o =>
o.ListenLocalhost(callbackPort, l => l.Protocols = HttpProtocols.Http2));
builder.Services.AddCaptureMcpTools(o =>
{
o.GatewayAddress = mock.GatewayAddress;
o.ApiBaseUrl = mock.ApiBaseUrl;
o.CallbackAddress = new Uri($"http://localhost:{callbackPort}");
o.ServiceName = "mysvc";
o.AppClientId = "mysvc-client";
o.AppClientSecret = "mysvc-secret";
}).WithToolsFromAssembly();
var app = builder.Build();
app.MapCaptureMcpTools();
await app.StartAsync();
// registration is a background retry loop: await it, do not assume it
var registration = await mock.WaitForRegistrationAsync();
registration.Tool("mysvc_greet").RequiredPermissions
.Should().BeEquivalentTo([$"{mock.AppKey}:greetings:send"]);
// call the tool as a caller who holds the permission at company 1
var alice = mock.CreateCaller().Grant(companyId: 1, "greetings:send");
var result = await mock.CallToolAsync("mysvc_greet", new { name = "World", companyId = 1 }, alice);
result.IsError.Should().BeFalse();
// and as one who does not
var bob = mock.CreateCaller();
var denied = await mock.CallToolAsync("mysvc_greet", new { name = "World", companyId = 1 }, bob);
denied.PermissionDenied.Should().BeTrue();
denied.RequiredPermissions.Should().Contain($"{mock.AppKey}:greetings:send");
Arguments are serialized in camelCase. An overload takes raw JSON and a raw token. Pass null for no token: that throws an RpcException with Unauthenticated, because the SDK refuses the call before your tool runs.
What the mock exposes
| Python | .NET | What it is for |
|---|---|---|
gateway_address, api_base_url | GatewayAddress, ApiBaseUrl | What 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_permissions | Registrations, UnregisteredServices, DeclaredPermissions | Everything 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_count | Python 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, withTool(name)to find one. - A tool (
MockRegisteredTool) has the name, description, input and output schema, the required permissions, whether it has annotations, andread_only,destructive,idempotentandopen_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
granttakes 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. Usegrant_namespacedto check against the permission of another app.- Caller tokens are unsigned and shaped like a JWT (
alg: none, with a realiss,subandexp). The SDK readsissandexpfrom 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.
- Python
- .NET
# 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", ...)
await using var mock = await CaptureMcpMockServer.StartAsync(o =>
{
o.GatewayPort = 6001;
o.ApiPort = 5001;
o.Host = "0.0.0.0"; // anything but "localhost" binds every interface
o.AdvertiseHost = "192.168.1.50"; // the LAN IP other machines dial: "0.0.0.0" itself is not routable
});
var caller = mock.CreateCaller().Grant(companyId: 1, "greetings:send");
Console.WriteLine($"gateway : {mock.GatewayAddress}");
Console.WriteLine($"api : {mock.ApiBaseUrl}");
Console.WriteLine($"token : {caller.Token}");
await Task.Delay(Timeout.Infinite);
Host controls what is bound, and AdvertiseHost controls what is published in GatewayAddress and ApiBaseUrl. They default to the same value, which works when Host is already routable, such as a hostname or a specific IP. A wildcard bind address is never dialable, so StartAsync throws if AdvertiseHost is not set next to one.
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.