AgentKeeper Agent Runtime SDK for Python

Package for sending promptless AI agent runtime events to AgentKeeper.

pip install --pre agentkeeper-runtime-sdk

The Python package supports custom agents, LangChain, LangGraph, AWS Bedrock Runtime through boto3-style clients, OpenAI Agents SDK, Azure OpenAI, Claude Managed Agents, and Anthropic SDK message/tool surfaces. Vercel AI SDK remains TypeScript-only.

Usage

import os
from agentkeeper_runtime_sdk import create_agentkeeper_runtime_client

ak = create_agentkeeper_runtime_client(
    endpoint="https://agentkeeper.dev",
    api_key=os.environ.get("AGENTKEEPER_RUNTIME_SDK_KEY"),
    runtime_service="support-agent",
    runtime_environment=os.environ.get("APP_ENV", "local"),
    runtime_integration="custom",
)

ak.track({
    "event_kind": "runtime_heartbeat",
    "capability": "observe",
    "evidence_summary": "AI agent runtime connected",
})
ak.flush()

Guarded Tools

def lookup_customer(args):
    return {"customer_id": args["customer_id"], "tier": "enterprise"}

guarded_lookup = ak.wrap_tool(
    "lookup_customer",
    lookup_customer,
    evaluate=lambda args: {"verdict": "passed", "reason": "allowed"},
)

guarded_lookup({"customer_id": "cus_123"})

OpenAI Agents SDK

Wrap Python functions before passing them to function_tool so the OpenAI Agents SDK can still infer the function signature.

import asyncio
import os
from agents import Agent, Runner, function_tool
from agentkeeper_runtime_sdk import create_agentkeeper_runtime_client
from agentkeeper_runtime_sdk.openai_agents import (
    wrap_openai_agents_run,
    wrap_openai_agents_tool,
)

ak = create_agentkeeper_runtime_client(
    api_key=os.environ.get("AGENTKEEPER_RUNTIME_SDK_KEY"),
    runtime_service="support-agent",
    runtime_environment=os.environ.get("APP_ENV", "local"),
    runtime_integration="openai_agents",
)

def lookup_customer(customer_id: str) -> dict:
    return {"customer_id": customer_id, "tier": "enterprise"}

guarded_lookup_customer = function_tool(
    wrap_openai_agents_tool(
        lookup_customer,
        ak,
        tool_name="lookup_customer",
        evaluate=lambda args: {"verdict": "passed"},
    )
)

async def main():
    guarded_run = wrap_openai_agents_run(Runner.run, ak)
    agent = Agent(name="Support triage", tools=[guarded_lookup_customer])
    await guarded_run(agent, "Check customer risk without exposing raw payloads.")
    ak.flush()

asyncio.run(main())

Anthropic SDK

Anthropic message calls are model-only observation. Wrap local tool handlers or runnable run(input) functions for blocking.

import os
from anthropic import Anthropic
from agentkeeper_runtime_sdk import create_agentkeeper_runtime_client
from agentkeeper_runtime_sdk.anthropic import (
    wrap_anthropic_client,
    wrap_anthropic_tool,
)

ak = create_agentkeeper_runtime_client(
    api_key=os.environ.get("AGENTKEEPER_RUNTIME_SDK_KEY"),
    runtime_service="support-agent",
    runtime_environment=os.environ.get("APP_ENV", "local"),
    runtime_integration="anthropic_sdk",
)

anthropic = wrap_anthropic_client(Anthropic(), ak)
lookup_customer = wrap_anthropic_tool(
    "lookup_customer",
    lambda args: {"customer_id": args["customer_id"], "tier": "enterprise"},
    ak,
    evaluate=lambda args: {"verdict": "passed"},
)

message = anthropic.messages.create(
    model=os.environ.get("ANTHROPIC_MODEL", "claude-sonnet-4-6"),
    max_tokens=1024,
    tools=[{
        "name": "lookup_customer",
        "description": "Look up a customer before support actions.",
        "input_schema": {
            "type": "object",
            "properties": {"customer_id": {"type": "string"}},
            "required": ["customer_id"],
        },
    }],
    messages=[{"role": "user", "content": "Check customer risk."}],
)

for block in getattr(message, "content", []):
    if getattr(block, "type", None) == "tool_use" and getattr(block, "name", None) == "lookup_customer":
        lookup_customer(getattr(block, "input", {}))

ak.flush()

Azure OpenAI

Azure OpenAI is model-only observation. The wrapper records safe metadata for chat.completions.create(), responses.create(), and embeddings.create() without storing prompts, messages, inputs, outputs, or provider request/response bodies.

import os
from openai import AzureOpenAI
from agentkeeper_runtime_sdk import create_agentkeeper_runtime_client
from agentkeeper_runtime_sdk.azure_openai import wrap_azure_openai_client

ak = create_agentkeeper_runtime_client(
    api_key=os.environ.get("AGENTKEEPER_RUNTIME_SDK_KEY"),
    runtime_service="support-agent",
    runtime_environment=os.environ.get("APP_ENV", "local"),
    runtime_integration="azure_openai",
)

deployment = os.environ.get("AZURE_OPENAI_DEPLOYMENT", "gpt-4o-mini")
azure_openai = wrap_azure_openai_client(
    AzureOpenAI(
        api_key=os.environ["AZURE_OPENAI_API_KEY"],
        azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
        api_version=os.environ.get("AZURE_OPENAI_API_VERSION", "2024-10-21"),
    ),
    ak,
)

azure_openai.chat.completions.create(
    model=deployment,
    messages=[{"role": "user", "content": "Check customer risk."}],
)

ak.flush()

Claude Managed Agents

Claude Managed Agents are hosted-session observation. The wrapper records safe agent, environment, session, event, stream, token, tool-name, and stop-reason metadata around Anthropic client.beta.* calls. It does not claim local pre-tool blocking for Anthropic-hosted sandbox/tool execution.

import os
from anthropic import Anthropic
from agentkeeper_runtime_sdk import create_agentkeeper_runtime_client
from agentkeeper_runtime_sdk.claude_managed_agents import wrap_claude_managed_agents_client

ak = create_agentkeeper_runtime_client(
    api_key=os.environ.get("AGENTKEEPER_RUNTIME_SDK_KEY"),
    runtime_service="support-agent",
    runtime_environment=os.environ.get("APP_ENV", "local"),
    runtime_integration="claude_managed_agents",
)

client = wrap_claude_managed_agents_client(Anthropic(), ak)
agent = client.beta.agents.create(
    name="Support triage",
    model=os.environ.get("ANTHROPIC_MANAGED_AGENT_MODEL", "claude-sonnet-4-6"),
    system="Help support engineers without exposing raw payloads.",
    tools=[{"type": "agent_toolset_20260401"}],
)

session = client.beta.sessions.create(
    agent=agent.id,
    environment_id=os.environ["ANTHROPIC_ENVIRONMENT_ID"],
    title="Support triage",
)

with client.beta.sessions.events.stream(session.id) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[{
            "type": "user.message",
            "content": [{"type": "text", "text": "Check customer risk."}],
        }],
    )
    for event in stream:
        if getattr(event, "type", None) == "session.status_idle":
            break

ak.flush()

LangChain and LangGraph

from agentkeeper_runtime_sdk import create_agentkeeper_runtime_client
from agentkeeper_runtime_sdk.langchain import (
    create_langchain_callback_handler,
    wrap_langchain_tool,
)

ak = create_agentkeeper_runtime_client(
    api_key=os.environ.get("AGENTKEEPER_RUNTIME_SDK_KEY"),
    runtime_service="support-agent",
    runtime_integration="langchain",
)

callbacks = [create_langchain_callback_handler(ak)]
guarded_lookup = wrap_langchain_tool(customer_lookup_tool, ak)

chain.invoke({"input": "Check customer risk."}, {"callbacks": callbacks})
ak.flush()

Use create_langgraph_callback_handler() and wrap_langgraph_tool() for LangGraph.

AWS Bedrock Runtime

import json
import boto3
from agentkeeper_runtime_sdk import create_agentkeeper_runtime_client
from agentkeeper_runtime_sdk.bedrock import wrap_bedrock_runtime_client

ak = create_agentkeeper_runtime_client(
    api_key=os.environ.get("AGENTKEEPER_RUNTIME_SDK_KEY"),
    runtime_service="support-agent",
    runtime_integration="bedrock",
)

bedrock = wrap_bedrock_runtime_client(
    boto3.client("bedrock-runtime", region_name="us-east-1"),
    ak,
)

bedrock.converse(
    modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
    messages=[{"role": "user", "content": [{"text": "Hello"}]}],
)

bedrock.invoke_model(
    modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
    body=json.dumps({
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 64,
        "messages": [{"role": "user", "content": "Hello"}],
    }),
)
ak.flush()

Privacy

Raw prompts, assistant output, tool arguments, provider request bodies, provider response bodies, and file contents are rejected by default. Send redacted summaries, hashes, model names, tool names, token counts, domains, and structured detector evidence instead.