Your agent,
ready to ship.

Installation

Install the SDK with pip or uv. Core depends only on Pydantic — extras add provider clients and the A2A server.

shellCopy
# Core (pydantic only)
pip install agentinc-sdk

# With OpenAI + A2A server
pip install "agentinc-sdk[openai,serve]"

# With Anthropic + A2A server
pip install "agentinc-sdk[anthropic,serve]"

# Everything
pip install "agentinc-sdk[all]"

Agent Skill

Install the agentinc-sdk skill so your coding agent understands the Agent API and can help you build agents faster.

shellCopy
npx skills add agentinc/sdk

Quickstart

Create an Agent, give it a model and a tool, then serve it over A2A.

quickstart.pyCopy
import os
from agentinc.sdk import Agent
from agentinc.sdk.serve import serve

def get_weather(city: str) -> str:
    """Gets the current weather for a city."""
    return f"72°F and sunny in {city}"

agent = Agent(
    role="You are a helpful assistant.",
    model={"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]},
    tools=[get_weather],
)

serve(agent, name="my-agent", port=8000)

shellCopy
# Start the server
python quickstart.py

# Call it
curl -X POST http://localhost:8000 \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"tasks/send","params":{"id":"t1","message":{"role":"user","parts":[{"type":"text","text":"What is the weather in Paris?"}]}}}'

Core Concepts

The SDK defines the contract between your agent code and the Agentinc platform.

Agent

The main developer-facing class. Implements AgentProtocol — pass it directly to serve().

pythonCopy
Agent(
    role:    str,                         # system prompt / persona
    model:   ModelConfig,                 # provider + credentials dict
    tools:   list[Callable] = [],         # plain functions — auto-wrapped
    mcps:    list[MCPConfig] = [],        # MCP server connections
    memory:  MemoryConfig | None = None,  # Redis-backed session memory
    context: str | None = None,           # extra context appended to system prompt
    data:    DataConfig | None = None,    # RAG config (reserved)
)

ModelConfig

Provider credentials as a plain dict. Provider auto-detected from model name prefix. When base_url is set, the OpenAI-compatible provider is used regardless of model name.

pythonCopy
# OpenAI
{"model": "gpt-4o-mini",       "api_key": "sk-..."}

# Anthropic
{"model": "claude-sonnet-4-6", "api_key": "sk-ant-..."}

# Google Gemini
{"model": "gemini-1.5-pro",    "api_key": "..."}

# Any OpenAI-compatible endpoint (DeepSeek, Groq, Ollama, …)
{"model": "deepseek-chat", "api_key": "sk-...", "base_url": "https://api.deepseek.com"}

MemoryConfig

Redis-backed session memory. History is loaded at the start of each run() and saved at the end. Requires pip install 'agentinc-sdk[memory]'.

pythonCopy
{
    "type":       "redis",
    "connection": "redis://localhost:6379/0",
    "user":       "optional",
    "password":   "optional",
}

Session ID from input.metadata["session_id"]. Falls back to a per-request UUID when absent. History stored with a 24-hour TTL.

shell — passing session_idCopy
curl -X POST http://localhost:8000 \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"tasks/send","params":{"id":"t1","metadata":{"session_id":"user-123"},"message":{"role":"user","parts":[{"type":"text","text":"My name is Alice"}]}}}'

MCPConfig

Connect to MCP servers. Tool schemas are fetched on the first run() call and merged with local tools. Requires pip install 'agentinc-sdk[mcp]'.

pythonCopy
# stdio — spawns a subprocess
{
    "type":    "stdio",
    "command": "npx",
    "args":    ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
}

# SSE — connects to a running HTTP server
{
    "type": "sse",
    "url":  "http://localhost:3001/sse",
}

AgentProtocol

The universal agent contract. Use this for framework integrations that manage their own LLM calls. Agent already implements it.

protocol.pyCopy
@runtime_checkable
class AgentProtocol(Protocol):
    def run(self, input: AgentInput) -> AsyncIterator[AgentOutput]: ...

Direct implementation

my_agent.pyCopy
from agentinc.sdk import AgentInput, AgentOutput, AgentProtocol
from agentinc.sdk.serve import serve

class MyAgent:
    async def run(self, input: AgentInput):
        yield AgentOutput(content=f"Hello! You said: {input.message}", done=True)

assert isinstance(MyAgent(), AgentProtocol)  # passes
serve(MyAgent(), name="my-agent", port=8000)

ToolProtocol

The contract for tools the agent can invoke.

protocol.pyCopy
@runtime_checkable
class ToolProtocol(Protocol):
    def schema(self) -> ToolSchema: ...
    async def call(self, tool_call: ToolCall) -> str: ...

Schemas

Pydantic models forming the data contract between agents, tools, and the platform.

@tool Decorator

Turn any function into a ToolProtocol with one line. Schema auto-generated from type hints and docstring.

pythonCopy
from agentinc.sdk import tool

# Bare — schema from type hints + docstring
@tool
async def search(query: str, limit: int = 10) -> str:
    """Searches the web for a query."""
    return f"Results for {query}"

# With explicit name / description
@tool(name="add", description="Adds two numbers")
def add_numbers(a: float, b: float) -> str:
    return str(a + b)

Type mappings

ToolWrapper

The object returned by @tool. Satisfies ToolProtocol — call directly or via a ToolCall.

pythonCopy
from agentinc.sdk import tool, ToolCall

@tool
def greet(name: str) -> str:
    """Says hello."""
    return f"Hello, {name}!"

# Via ToolCall (how the platform dispatches)
result = await greet.call(ToolCall(id="1", name="greet", arguments={"name": "World"}))

# Direct call
result = await greet(name="World")

RawAdapter ⚠ Deprecated

Kept for backward compatibility. Emits DeprecationWarning on instantiation. Will be removed in v0.3.

A2A Server

Serve any AgentProtocol as an A2A-compliant HTTP endpoint. Requires [serve] extra.

create_app()

pythonCopy
from agentinc.sdk import Agent
from agentinc.sdk.serve import create_app

app = create_app(
    Agent(role="You are helpful.", model={"model": "gpt-4o-mini", "api_key": "sk-..."}),
    name="my-agent",
)
# uvicorn myfile:app --port 8000

serve()

Example: OpenAI Agent

Minimal Agent with a tool using an OpenAI model.

openai_agent.pyCopy
import os
from agentinc.sdk import Agent
from agentinc.sdk.serve import serve

def get_weather(city: str) -> str:
    """Gets the current weather for a city."""
    return f"72°F and sunny in {city}"

agent = Agent(
    role="You are a helpful assistant.",
    model={"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]},
    tools=[get_weather],
)

serve(agent, name="openai-agent", port=8000)

Example: Anthropic Agent

Claude-powered agent — provider auto-detected from the claude- prefix.

anthropic_agent.pyCopy
import os
from agentinc.sdk import Agent
from agentinc.sdk.serve import serve

def lookup(topic: str) -> str:
    """Looks up a fact in the knowledge base."""
    return f"Here's what we know about {topic}: it's fascinating."

agent = Agent(
    role="You are a helpful assistant.",
    model={"model": "claude-sonnet-4-6", "api_key": os.environ["ANTHROPIC_API_KEY"]},
    tools=[lookup],
)

serve(agent, name="anthropic-agent", port=8002)

Example: Session Memory

Persistent multi-turn conversations backed by Redis.

memory_agent.pyCopy
import os
from agentinc.sdk import Agent
from agentinc.sdk.serve import serve

agent = Agent(
    role="You are a friendly assistant with a long memory.",
    model={"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]},
    memory={
        "type":       "redis",
        "connection": os.environ.get("REDIS_URL", "redis://localhost:6379/0"),
        "password":   os.environ.get("REDIS_PASSWORD"),
    },
)

serve(agent, name="memory-agent", port=8000)

shell — two-turn session testCopy
# Turn 1
curl -X POST http://localhost:8000 \
  -d '{"jsonrpc":"2.0","id":1,"method":"tasks/send","params":{"id":"t1","metadata":{"session_id":"demo"},"message":{"role":"user","parts":[{"type":"text","text":"My name is Alice"}]}}}'

# Turn 2 — agent recalls the name
curl -X POST http://localhost:8000 \
  -d '{"jsonrpc":"2.0","id":2,"method":"tasks/send","params":{"id":"t2","metadata":{"session_id":"demo"},"message":{"role":"user","parts":[{"type":"text","text":"What is my name?"}]}}}'

Example: MCP Server

Connect to an MCP server. Tool schemas are fetched on first call and merged with local tools.

mcp_agent.pyCopy
import os
from agentinc.sdk import Agent
from agentinc.sdk.serve import serve

def read_file(path: str) -> str:
    """Reads a local file."""
    with open(path) as f:
        return f.read()

agent = Agent(
    role="You are a helpful file assistant.",
    model={"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]},
    tools=[read_file],   # local tools
    mcps=[{              # MCP tools — discovered on first run()
        "type":    "stdio",
        "command": "npx",
        "args":    ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
    }],
)

serve(agent, name="mcp-agent", port=8000)

Example: LangChain Agent

LangChain manages its own LLM calls — implement AgentProtocol directly.

langchain_agent.pyCopy
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langchain_core.tools import tool as lc_tool
from langchain_openai import ChatOpenAI
from agentinc.sdk import AgentInput, AgentOutput
from agentinc.sdk.serve import serve

@lc_tool
def calculator(expression: str) -> str:
    """Evaluates a math expression."""
    return str(eval(expression))

llm = ChatOpenAI(model="gpt-4o-mini").bind_tools([calculator])

class LangChainAgent:
    async def run(self, input: AgentInput):
        messages = [SystemMessage(content="You are a helpful assistant.")]
        for msg in input.history:
            messages.append(
                HumanMessage(content=msg.content) if msg.role == "user"
                else AIMessage(content=msg.content or "")
            )
        messages.append(HumanMessage(content=input.message))
        response = await llm.ainvoke(messages)
        if response.tool_calls:
            yield AgentOutput(
                tool_calls=[{"id": tc["id"], "name": tc["name"], "arguments": tc["args"]}
                            for tc in response.tool_calls],
                done=False,
            )
        else:
            yield AgentOutput(content=response.content, done=True)

serve(LangChainAgent(), name="langchain-agent", port=8003)

Example: CrewAI Agent

CrewAI manages its own orchestration — implement AgentProtocol directly and wrap the blocking call.

crewai_agent.pyCopy
import asyncio
from crewai import Agent as CrewAgent, Crew, Task
from agentinc.sdk import AgentInput, AgentOutput
from agentinc.sdk.serve import serve

researcher = CrewAgent(
    role="Senior Research Analyst",
    goal="Provide thorough research on any topic",
    backstory="You are an experienced research analyst.",
    verbose=False,
)

class CrewAIAgent:
    async def run(self, input: AgentInput):
        task = Task(description=input.message, expected_output="Clear analysis", agent=researcher)
        crew = Crew(agents=[researcher], tasks=[task], verbose=False)
        # kickoff() is synchronous — wrap to avoid blocking the event loop
        result = await asyncio.to_thread(crew.kickoff)
        yield AgentOutput(content=str(result), done=True)

serve(CrewAIAgent(), name="crewai-agent", port=8004)

Ship to the Marketplace

Once your agent works locally, publish it to the Agentinc marketplace so tenants can discover and deploy it into their own isolated namespaces.

agent.json manifest

agent.jsonCopy
{
  "name":        "my-agent",
  "version":     "1.0.0",
  "description": "A helpful assistant with tool support",
  "entrypoint":  "agent.py",
  "runtime":     "python3.12",
  "capabilities": {
    "streaming": true,
    "tools":     true,
    "memory":    false
  }
}

Publish

shellCopy
# Login to the Agentinc registry
agentinc login

# Publish from your project directory
agentinc publish

# Publish a specific version
agentinc publish --version 1.2.0

# List your published agents
agentinc agents list

Entry point structure

Your entrypoint file must expose a top-level variable named agent that satisfies AgentProtocol.

agent.pyCopy
# agent.py — this file is what the platform imports
import os
from agentinc.sdk import Agent

def get_weather(city: str) -> str:
    """Gets the current weather for a city."""
    return f"72°F and sunny in {city}"

# The platform reads this variable — name must be "agent"
agent = Agent(
    role="You are a helpful weather assistant.",
    model={"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]},
    tools=[get_weather],
)

# Optional: run locally for testing
if __name__ == "__main__":
    from agentinc.sdk.serve import serve
    serve(agent, name="weather-agent", port=8000)

All Exports

Everything exported from agentinc.sdk.

pythonCopy
from agentinc.sdk import (
    # Core
    Agent,          # Main class — declare and run agents
    AgentProtocol,  # Protocol — implement run() for custom integrations
    ToolProtocol,   # Protocol — implement schema() + call()
    AgentFactory,   # Type alias — callable returning AgentProtocol
    # Schemas
    AgentInput,     # Model — agent invocation input
    AgentOutput,    # Model — single output chunk
    Message,        # Model — conversation history entry
    ToolCall,       # Model — tool invocation request
    ToolSchema,     # Model — tool JSON schema
    # Config TypedDicts
    ModelConfig,    # Provider + credentials
    MemoryConfig,   # Redis memory config
    MCPConfig,      # MCP server connection config
    DataConfig,     # RAG config (reserved)
    # Tools
    ToolWrapper,    # Class — wraps callable as ToolProtocol
    tool,           # Decorator — function → ToolWrapper
    # Deprecated
    RawAdapter,     # DEPRECATED — use Agent() instead
)

# A2A serve module (requires [serve] extra)
from agentinc.sdk.serve import create_app, serve