AgentClose Python

Python SDK and read-only MCP server for AgentClose post-payment reconciliation.

Install

pip install agentclose

Install the MCP extra when exposing AgentClose to an MCP client:

pip install "agentclose[mcp]"

Capture events

from agentclose import AgentCloseClient, Event

with AgentCloseClient("acl_your_api_key") as client:
    client.capture(
        Event(
            type="request.started",
            data={
                "agent_id": "research-agent",
                "task_id": "task_8f2",
                "request_id": "req_4a81",
                "service": "company-data.api",
                "attempt_number": 1,
            },
        )
    )

event_id and occurred_at are generated once when Event is constructed. A retry therefore keeps the same idempotency identity. Capture methods retry transient failures with a bounded backoff and are fail-open by default, so reconciliation telemetry does not break the Agent workflow. Set fail_open=False when delivery must raise an error.

Send 1–100 facts with one request:

events = [
    Event(type="task.started", data={"agent_id": "research-agent", "task_id": "task_8f2"}),
    Event(
        type="payment.confirmed",
        data={
            "agent_id": "research-agent",
            "task_id": "task_8f2",
            "request_id": "req_4a81",
            "payment_id": "pay_72bc",
            "amount": {"minor": 12, "currency": "USD"},
        },
    ),
]

with AgentCloseClient("acl_your_api_key") as client:
    client.capture_batch(events)

AsyncAgentCloseClient provides the same methods for async applications:

from agentclose import AsyncAgentCloseClient, Event

async with AsyncAgentCloseClient("acl_your_api_key") as client:
    await client.capture(Event(type="task.started", data={"task_id": "task_8f2"}))

Observe x402

The observer records an x402 exchange after your runtime has handled payment. It never holds funds, signs a payment, or reads wallet keys.

from agentclose import AgentCloseClient, X402Observation

observation = X402Observation(
    agent_id="research-agent",
    task_id="task_8f2",
    request_id="req_4a81",
    service="company-data.api",
    endpoint="https://company-data.example/query",
    payment_required="base64-payment-required-header",
    payment_response="base64-payment-response-header",
    http_status=200,
    delivery_observed=True,
)

with AgentCloseClient("acl_your_api_key") as client:
    client.observe_x402(observation)

Query reconciliation data

Read methods raise errors instead of failing open:

with AgentCloseClient("acl_your_api_key") as client:
    page = client.list_transactions(status="EXCEPTION", limit=50)
    next_cursor = page["next_cursor"]
    payment_matches = client.get_payment_transactions("pay_72bc")
    usage = client.get_usage()

Run the read-only MCP server

export AGENTCLOSE_API_KEY=acl_your_api_key
agentclose-mcp

The stdio server exposes these tools:

• list_transactions
• get_transaction
• get_task_transactions
• get_task_cost
• list_open_exceptions
• get_payment_status
• get_usage_summary

Example MCP client configuration:

{
  "mcpServers": {
    "agentclose": {
      "command": "agentclose-mcp",
      "env": {
        "AGENTCLOSE_API_KEY": "acl_your_project_api_key"
      }
    }
  }
}

The MCP process uses the same project-scoped AgentClose API key as the SDK. It only calls read endpoints and cannot capture events, resolve exceptions, rotate keys, or modify data.