Gt8004 SDK
Official Python SDK for GT8004 β AI agent observability with MCP, A2A, x402 payment tracking. FastAPI, Flask, FastMCP middleware included.
Ask AI about Gt8004 SDK
Powered by Claude Β· Grounded in docs
I know everything about Gt8004 SDK. Ask me about installation, configuration, usage, or troubleshooting.
0/500
Reviews
Documentation
GT8004 Python SDK
Official Python SDK for GT8004 - AI Agent Analytics & Observability Platform.
Installation
pip install git+https://github.com/vataops/gt8004-sdk.git
With framework-specific extras:
# FastAPI / Starlette
pip install "gt8004-sdk[fastapi] @ git+https://github.com/vataops/gt8004-sdk.git"
# FastMCP (MCP servers)
pip install "gt8004-sdk[mcp] @ git+https://github.com/vataops/gt8004-sdk.git"
# All frameworks
pip install "gt8004-sdk[all] @ git+https://github.com/vataops/gt8004-sdk.git"
Quick Start
MCP Server (FastMCP)
from fastmcp import FastMCP
from gt8004 import GT8004Logger
from gt8004.middleware.mcp import GT8004MCPMiddleware
logger = GT8004Logger(agent_id="your-agent-id", api_key="your-api-key", protocol="mcp")
logger.transport.start_auto_flush()
mcp = FastMCP("my-server")
mcp.add_middleware(GT8004MCPMiddleware(logger))
@mcp.tool()
def search(query: str) -> str:
return "results..."
# Automatically logs tool_name="search", protocol="mcp"
A2A Server (FastAPI)
from fastapi import FastAPI
from gt8004 import GT8004Logger
from gt8004.middleware.fastapi import GT8004Middleware
logger = GT8004Logger(agent_id="your-agent-id", api_key="your-api-key", protocol="a2a")
logger.transport.start_auto_flush()
app = FastAPI()
app.add_middleware(GT8004Middleware, logger=logger)
# Automatically extracts skill_id from A2A request bodies
@app.on_event("shutdown")
async def shutdown():
await logger.close()
A2A + x402 Payment (ASGI)
When using x402 payment middleware, use GT8004ASGIMiddleware as the outermost wrapper to capture all requests including 402 payment responses:
from fastapi import FastAPI
from gt8004 import GT8004Logger
from gt8004.middleware.asgi import GT8004ASGIMiddleware
logger = GT8004Logger(agent_id="your-agent-id", api_key="your-api-key", protocol="a2a")
logger.transport.start_auto_flush()
app = FastAPI()
# Apply x402 payment middleware first, then GT8004 outermost
app = x402_middleware(app)
app = GT8004ASGIMiddleware(app, logger)
# Automatically captures x402 payment fields (amount, tx_hash, payer)
Flask / Django
from flask import Flask
from gt8004 import GT8004Logger
from gt8004.middleware.flask import GT8004FlaskMiddleware
logger = GT8004Logger(agent_id="your-agent-id", api_key="your-api-key")
logger.transport.start_auto_flush()
app = Flask(__name__)
app.wsgi_app = GT8004FlaskMiddleware(app.wsgi_app, logger)
Your analytics are now live at https://gt8004.xyz/agents/{agent-id} with protocol-specific breakdowns.
Configuration
Environment Variables
export GT8004_AGENT_ID="your-agent-id"
export GT8004_API_KEY="your-api-key"
import os
from gt8004 import GT8004Logger
logger = GT8004Logger(
agent_id=os.environ["GT8004_AGENT_ID"],
api_key=os.environ["GT8004_API_KEY"],
network="mainnet", # "mainnet" (default) or "testnet"
protocol="a2a", # "mcp" or "a2a" (optional)
batch_size=50, # Flush after N entries (default: 50)
flush_interval=5.0, # Auto-flush interval in seconds (default: 5.0)
)
Lifecycle
# Start auto-flush background task
logger.transport.start_auto_flush()
# Verify connectivity on startup
ok = await logger.verify_connection()
# Graceful shutdown (flushes pending logs)
await logger.close()
Features
- Multi-framework support (FastMCP, FastAPI, Flask, Django)
- Protocol-aware logging (MCP, A2A)
- Automatic tool/skill name extraction per protocol
- x402 payment tracking (amount, tx hash, payer)
- Non-blocking async transport with batch buffering
- Auto-retry with exponential backoff
- Circuit breaker protection
Supported Frameworks
| Framework | Middleware | Install Extra |
|---|---|---|
| FastMCP (MCP servers) | GT8004MCPMiddleware | [mcp] |
| FastAPI / Starlette | GT8004Middleware | [fastapi] |
| Pure ASGI (x402 compatible) | GT8004ASGIMiddleware | [fastapi] |
| Flask / Django (WSGI) | GT8004FlaskMiddleware | (none) |
Excluding Paths
By default, health check endpoints are excluded from logging. You can customize this with exclude_paths:
# FastAPI / Starlette
app.add_middleware(GT8004Middleware, logger=logger, exclude_paths={"/health", "/metrics", "/internal"})
# Pure ASGI
app = GT8004ASGIMiddleware(app, logger, exclude_paths={"/health", "/metrics"})
Default excluded paths:
GT8004Middleware:/health,/healthz,/readyz,/_health,/.well-known/agent.json/healthGT8004ASGIMiddleware:/health,/healthz,/readyz,/_health
Protocol Support
| Protocol | Tool Name Source | Example |
|---|---|---|
| (none) | URL path last segment | /api/search -> search |
mcp | FastMCP on_call_tool hook | @mcp.tool() def search(...) -> search |
a2a | Request body skill_id or path | {"skill_id":"translate"} -> translate |
x402 Payment Logging
All middleware adapters automatically extract x402 payment data from HTTP headers:
x402_amount- Payment amount in USDCx402_tx_hash- On-chain settlement transaction hashx402_token- Token identifier (e.g.USDC-base-mainnet)x402_payer- Payer wallet address
No additional configuration needed β payment headers (X-Payment, X-Payment-Response) are parsed automatically when present.
License
MIT - see LICENSE
