agentsuite-sdk

AgentSuite SDK

Python SDK for connecting AI agent frameworks to the AgentSuite Gateway. Provides drop-in adapters for OpenAI Agents SDK, Google ADK, Claude Agent SDK, and LangChain with automatic session tracking and user query injection.

Installation

pip install agentsuite-sdk

With framework-specific extras:

pip install agentsuite-sdk[openai]    # OpenAI Agents SDK
pip install agentsuite-sdk[adk]       # Google ADK
pip install agentsuite-sdk[claude]    # Claude Agent SDK
pip install agentsuite-sdk[langchain] # LangChain (create_agent API)
pip install agentsuite-sdk[all]       # All frameworks

OpenAI Agents SDK

from agents import Agent, Runner, RunConfig
from agentsuite import GatewayClient

client = GatewayClient(url="...", api_key="sk-vai-...")
adapter = client.openai()

mcp_server = adapter.mcp_server()
await mcp_server.connect()

try:
    agent = Agent(
        name="my-agent",
        model="gpt-4o",
        mcp_servers=[mcp_server],
    )
    run_config = RunConfig(
        model="gpt-4o",
        call_model_input_filter=adapter.input_filter,
    )
    result = await Runner.run(agent, input="What are my open tickets?", run_config=run_config)
finally:
    await mcp_server.cleanup()

Google ADK

from google.adk.agents import Agent
from agentsuite import GatewayClient

client = GatewayClient(url="...", api_key="sk-vai-...")
adapter = client.adk()

agent = Agent(
    name="my_agent",
    model="gemini-2.0-flash",
    tools=[adapter.toolset()],
    before_model_callback=adapter.before_model_callback,
    before_tool_callback=adapter.before_tool_callback,
    after_tool_callback=adapter.after_tool_callback,
)

LangChain

from langchain.agents import create_agent
from langchain_core.messages import HumanMessage
from agentsuite import GatewayClient

client = GatewayClient(url="...", api_key="sk-vai-...")
adapter = client.langchain()

async with adapter:
    tools = await adapter.get_tools()
    graph = create_agent(
        model="openai:gpt-4o",
        tools=tools,
        system_prompt="You are a helpful assistant.",
    )
    result = await graph.ainvoke(
        {"messages": [HumanMessage(content=user_message)]},
        config={"callbacks": [adapter.create_callback()]},
    )

Claude Agent SDK

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient
from agentsuite import GatewayClient

client = GatewayClient(url="...", api_key="sk-vai-...")
adapter = client.claude()

options = ClaudeAgentOptions(
    **adapter.options(),
    model="claude-sonnet-4-5",
)

async with ClaudeSDKClient(options=options) as sdk:
    user_message = "What are my open tickets?"

    # Must call record_query() before sdk.query() — Claude has no before_model_callback
    adapter.record_query(user_message)
    await sdk.query(user_message)

    async for event in sdk.receive_response():
        ...

Direct Connection (no adapter)

Connect directly to the MCP Gateway without the adapter. The LLM fills in session_id and user_queries from the tool schema — simpler setup but less reliable query capture.

from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient, HookMatcher

options = ClaudeAgentOptions(
    mcp_servers={
        "gateway": {
            "type": "http",
            "url": "https://...",
            "headers": {"X-API-Key": "sk-vai-..."},
        }
    },
    hooks={
        "PreToolUse": [HookMatcher(matcher="^mcp__", hooks=[pre_tool_use_hook])],
        "PostToolUse": [HookMatcher(matcher="^mcp__", hooks=[post_tool_use_hook])],
    },
    permission_mode="bypassPermissions",
    setting_sources=[],
    model="claude-sonnet-4-5",
)

How It Works

Each adapter integrates the gateway by hooking into its framework's native extension points. All three share the same two-phase session pattern:

1. Before tool call — inject session_id (or "sess_new" on first call) + user_queries into args
2. After tool call — parse "Session ID: sess_xxxx" from the response, store it, and flush the query buffer

The difference is just where each framework lets you hook into that flow.

Session State & Deduplication

GatewaySession tracks which user messages have been sent to the gateway using a position watermark:

• _consumed_count — number of user messages already sent
• _pending — messages buffered since the last tool call

On every LLM call, the adapter extracts the full list of user messages from conversation history and calls process_history(). Only messages at index ≥ _consumed_count + len(_pending) are added to _pending. After a successful tool call, flush() advances _consumed_count and clears _pending.

This approach:

• Is immune to Python object reconstruction (which broke id()-based dedup in ADK)
• Allows repeated identical messages (which text-based dedup would silently drop)
• Works reliably as long as conversation history is append-only (guaranteed by all frameworks)

OpenAI Agents SDK

The SDK exposes Agent(mcp_servers=[...]) and RunConfig(call_model_input_filter=...) as the main extension points. Unlike ADK, it has no before_model_callback or pre-tool-call callback — so there is no hook to intercept tool calls before they execute. To work around this, the adapter wraps the SDK's native MCPServerStreamableHttp in a proxy class (GatewayMCPServer) that intercepts call_tool directly.

adapter.mcp_server() returns the proxy — it injects session context before every tool call and parses the gateway response to capture the assigned session_id.

adapter.input_filter is registered as call_model_input_filter. It fires before each LLM call and passes the full user-message history to process_history(), which applies the position watermark to capture only genuinely new messages.

Google ADK

ADK has first-class before_model_callback, before_tool_callback, and after_tool_callback hooks on Agent — exactly what's needed, with no wrapping required.

adapter.toolset() returns a native MCPToolset. The three callbacks slot directly into Agent(...):

• before_model_callback extracts user messages from llm_request.contents and passes them to process_history()
• before_tool_callback injects session_id and user_queries into MCP tool args
• after_tool_callback parses the session ID from the response and calls flush()

Tool callbacks are scoped to McpTool instances only — regular function tools are ignored.

LangChain

The adapter uses the new create_agent API and maps onto the gateway's needs via a callback and a tool wrapper:

• adapter.create_callback() — returns a callback whose on_chain_start fires once per ainvoke() at run start. Used to capture user messages from inputs["messages"] and buffer them into user_queries (same watermark logic as other adapters). Pass it in config={"callbacks": [adapter.create_callback()]} so you don't need to call record_user_message().
• Tool wrapper (applied to every tool from get_tools()) — runs before and after each gateway tool call. Before: injects session_id and user_queries into the tool arguments. After: extracts the gateway-assigned session_id from the response and flushes the buffer.

adapter.record_user_message(message) is the alternative when you are not using the callback (e.g. custom input shape or you prefer explicit control).

Claude Agent SDK

The Claude SDK wraps the Claude CLI subprocess and exposes PreToolUse/PostToolUse hooks. It has no before_model_callback, so user messages cannot be captured automatically.

adapter.options() returns a dict for ClaudeAgentOptions(...) that wires up the MCP server, hooks, permission_mode="bypassPermissions", and setting_sources=[] (to isolate from ~/.claude settings).

adapter.record_query(message) must be called manually before each sdk.query() call. This is an intentional trade-off — the alternative is the Direct Connection path (no adapter), where the LLM fills in session_id and user_queries itself from the tool schema. That requires no SDK at all, but is less reliable since it depends on the LLM correctly tracking session state.

Action Guard SDK

The Guard SDK provides a standalone client for evaluating tool calls against security policies before execution. It follows OpenAI SDK patterns with shared HTTP clients, context managers, and typed exceptions.

Basic Usage

from agentsuite import ActionGuardClient, GuardError

# Sync client with context manager (recommended)
with ActionGuardClient(api_key="sk-vai-...", policy_id="agp_...") as guard:
    result = guard.actions.guard_query(
        query="Tool: delete_file, Args: {'path': '/etc/passwd'}",
    )

    if not result.allowed:
        print(f"Blocked: {result.explanation}")
        print(f"Violations: {result.violations}")

Async Client

from agentsuite import AsyncActionGuardClient

# Async client with context manager
async with AsyncActionGuardClient() as guard:
    result = await guard.actions.guard_query(query="...")

    if not result.allowed:
        raise PermissionError(result.explanation)

Error Handling

In a guardrail hook, the only real decision is fail closed (deny on error, safer) or fail open (allow on error). Catching GuardError is all you need:

from agentsuite import AsyncActionGuardClient, GuardError

async def action_guard_hook(...):
    try:
        result = await guard.actions.guard_query(query=f"Tool: {tool_name}, Args: {tool_args}")
        if not result.allowed:
            return reject(result.explanation)
        return allow()
    except GuardError:
        # Guard is unreachable — fail closed (deny) to stay safe
        return reject("Action Guard unavailable")

Startup validation — ActionGuardClient() raises GuardConfigError if required env vars are missing. Catch this at boot before your agent starts:

from agentsuite.guard import GuardConfigError

try:
    guard = ActionGuardClient()  # raises GuardConfigError if VIRTUE_API_KEY or ACTION_GUARD_POLICY_ID not set
except GuardConfigError as e:
    raise SystemExit(f"Guard misconfigured: {e}")

Observability — use status_code to log different failure modes in production. Only GuardAPIStatusError carries a status code; catch it before the base GuardError:

from agentsuite.guard import GuardAPIStatusError, GuardError

except GuardAPIStatusError as e:
    if e.status_code == 401:
        logger.critical("Guard auth failed — check VIRTUE_API_KEY")
    elif e.status_code == 503:
        logger.warning("Guard service unavailable")
    else:
        logger.error("Guard API error: %s (status=%s)", e.message, e.status_code)
    return reject("Guard unavailable")
except GuardError as e:
    logger.error("Guard connection error: %s", e.message)
    return reject("Guard unavailable")

Stateful vs Stateless

Stateless (guard_query): Evaluate a single query with no context.

result = guard.actions.guard_query(
    query="Tool: delete_user, Args: {'username': 'admin'}",
    fast_mode=True,  # Lower latency
)

Stateful (guard): Pass full session history for context-aware decisions.

result = guard.actions.guard(
    session_id="user-123",
    session_history={
        "session_info": {"session_id": "user-123"},
        "trajectory": [...],
    },
)

Resource Management

The SDK automatically manages HTTP connections. Use context managers for automatic cleanup:

# Automatic cleanup (recommended)
with ActionGuardClient() as guard:
    result = guard.actions.guard_query(query="...")

# Manual cleanup
guard = ActionGuardClient()
try:
    result = guard.actions.guard_query(query="...")
finally:
    guard.close()  # Or await guard.aclose() for async

Custom HTTP Client

For advanced use cases, provide your own httpx.Client:

import httpx

# Sync
with httpx.Client(timeout=60.0, limits=httpx.Limits(max_connections=100)) as http:
    guard = ActionGuardClient(http_client=http)
    result = guard.actions.guard_query(query="...")

# Async
async with httpx.AsyncClient(timeout=60.0) as http:
    guard = AsyncActionGuardClient(http_client=http)
    result = await guard.actions.guard_query(query="...")

Configuration

Enable debug logging programmatically:

import logging
logging.getLogger("agentsuite").setLevel(logging.DEBUG)

License

MIT