Your agent,
ready to ship.

Installation

Install the SDK with pip or uv, from GitHub: it is not on PyPI yet. Core depends only on Pydantic — extras add provider clients and the A2A server. Pin a tag or commit instead of @dev for a fixed version.

shellCopy
# Core (pydantic only)
pip install "genfleet-sdk @ git+https://github.com/genfleet/sdk@dev"

# With OpenAI + A2A server
pip install "genfleet-sdk[openai,serve] @ git+https://github.com/genfleet/sdk@dev"

# With Anthropic + A2A server
pip install "genfleet-sdk[anthropic,serve] @ git+https://github.com/genfleet/sdk@dev"

# Everything
pip install "genfleet-sdk[all] @ git+https://github.com/genfleet/sdk@dev"

Agent Skill

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

shellCopy
npx skills add genfleet/sdk

Quickstart

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

quickstart.pyCopy
import os
from genfleet.sdk import Agent
from genfleet.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": "openai/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" -H 'A2A-Version: 1.0' \
  -d '{"jsonrpc":"2.0","id":1,"method":"SendMessage","params":{"message":{"messageId":"m1","role":"ROLE_USER","parts":[{"text":"What is the weather in Paris?"}]}}}'

Core Concepts

The SDK defines the contract between your agent code and the Genfleet 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 | "platform" | None = None,  # episodic memory (platform store or Redis)
    context: str | None = None,           # extra context appended to system prompt
    audit:   AuditConfig | None = None,   # structured audit logging
)

ModelConfig

Provider credentials as a plain dict. Use the provider/model-name format to specify the provider explicitly.

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

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

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

# Any OpenAI-compatible endpoint (DeepSeek, Groq, Ollama, …)
{"model": "openai/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 'genfleet-sdk[memory] @ git+https://github.com/genfleet/sdk@dev'.

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" -H 'A2A-Version: 1.0' \
  -d '{"jsonrpc":"2.0","id":1,"method":"SendMessage","params":{"message":{"messageId":"m1","role":"ROLE_USER","contextId":"user-123","parts":[{"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 'genfleet-sdk[mcp] @ git+https://github.com/genfleet/sdk@dev'.

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",
}

AuditConfig

Opt-in structured audit logging. Captures invocation lifecycle, LLM calls, tool dispatch, token usage, and errors. No overhead when audit=None (the default).

python — console backendCopy
agent = Agent(
    role="You are a helpful assistant.",
    model={"model": "openai/gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]},
    audit={"backend": "console", "agent_name": "my-agent"},
)

python — file backend (JSONL)Copy
audit={"backend": "file", "file_path": "audit.jsonl"}

python — callback backendCopy
async def my_handler(event):
    print(event.event_type, event.data)

audit={"backend": "callback", "callback": my_handler}

Audit Events

Token usage (input/output/total tokens) is tracked automatically for OpenAI, Anthropic, and Gemini providers.

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 genfleet.sdk import AgentInput, AgentOutput, AgentProtocol
from genfleet.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 genfleet.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 genfleet.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")

A2A Server

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

create_app()

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

app = create_app(
    Agent(role="You are helpful.", model={"model": "openai/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 genfleet.sdk import Agent
from genfleet.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": "openai/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 genfleet.sdk import Agent
from genfleet.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": "anthropic/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 genfleet.sdk import Agent
from genfleet.sdk.serve import serve

agent = Agent(
    role="You are a friendly assistant with a long memory.",
    model={"model": "openai/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":"SendMessage","params":{"message":{"messageId":"m1","role":"ROLE_USER","contextId":"demo","parts":[{"text":"My name is Alice"}]}}}'

# Turn 2 — agent recalls the name
curl -X POST http://localhost:8000 \
  -d '{"jsonrpc":"2.0","id":2,"method":"SendMessage","params":{"message":{"messageId":"m2","role":"ROLE_USER","contextId":"demo","parts":[{"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 genfleet.sdk import Agent
from genfleet.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": "openai/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 genfleet.sdk import AgentInput, AgentOutput
from genfleet.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 genfleet.sdk import AgentInput, AgentOutput
from genfleet.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)

Publish on Genfleet (private beta)

From an empty directory to chatting with your own agent on the platform: install, write an agent with a tool, test it locally, push it, run it in your workspace, chat. Every command here was checked against the SDK and the gf CLI on 2026-09-30.

1. Install

Python 3.12 or newer and uv. The SDK and the CLI are installed from GitHub; neither is on PyPI yet.

shellCopy
# the gf CLI (log in, push agents)
uv tool install git+https://github.com/genfleet/cli
gf --help

export GENFLEET_API=https://…   # the platform URL from your invitation

2. Write an agent with a tool

An agent is a directory with a Python package inside it. The manifest, genfleet.toml, sits inside the package, next to agent.py:

layoutCopy
hello-agent/
  pyproject.toml
  hello_agent/
    __init__.py
    genfleet.toml     # manifest: name, kind, entry
    agent.py          # build_agent() factory + the tool
    __main__.py       # serve it locally
  tests/
    test_agent.py

hello-agent/pyproject.tomlCopy
[project]
name = "hello-agent"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = ["genfleet-sdk[openai,serve] @ git+https://github.com/genfleet/sdk@dev"]

[dependency-groups]
dev = ["pytest"]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.metadata]
allow-direct-references = true

[tool.hatch.build.targets.wheel]
packages = ["hello_agent"]

hello-agent/hello_agent/genfleet.tomlCopy
[agent]
name = "hello-agent"
kind = "code"
description = "Says hello and counts words."
entry = "hello_agent.agent:build_agent"

hello-agent/hello_agent/agent.pyCopy
import os

from genfleet.sdk import Agent, tool

# `openrouter/` is the SDK's OpenRouter prefix: one key, any vendor's model.
# A `:free` model costs nothing; free ids rotate, so swap in another `:free`
# model that supports tool calling if this one stops resolving.
MODEL = "openrouter/nvidia/nemotron-3.5-lightning:free"


@tool(description="Counts the words in a piece of text.")
def count_words(text: str) -> str:
    return str(len(text.split()))


def build_agent(api_key: str | None = None) -> Agent:
    # A factory, not a module-level Agent(): the package stays importable
    # without a key, so tests can build it with a dummy one.
    key = api_key or os.environ.get("OPENROUTER_API_KEY")
    if not key:
        raise RuntimeError("Set OPENROUTER_API_KEY (on the platform: a secret, see step 5).")
    return Agent(
        role="You are a friendly assistant. Use count_words when asked to count words.",
        model={"model": MODEL, "api_key": key},
        tools=[count_words],
    )

hello-agent/hello_agent/__init__.py and __main__.pyCopy
# __init__.py
from .agent import build_agent

__all__ = ["build_agent"]

# __main__.py
from genfleet.sdk.serve import serve

from .agent import build_agent

if __name__ == "__main__":
    serve(build_agent(), name="hello-agent", port=8000)

3. Test it locally

A test that needs no network and no key:

hello-agent/tests/test_agent.pyCopy
import asyncio

from genfleet.sdk import ToolCall
from hello_agent.agent import build_agent, count_words


def test_builds_without_a_real_key():
    assert build_agent(api_key="test") is not None


def test_count_words():
    call = ToolCall(id="1", name="count_words", arguments={"text": "one two three"})
    assert asyncio.run(count_words.call(call)) == "3"

shellCopy
cd hello-agent
uv sync
uv run pytest -q                       # 2 passed

export OPENROUTER_API_KEY=sk-or-v1-…   # a free OpenRouter key works
uv run python -m hello_agent           # A2A server on :8000

In another shell:

shellCopy
curl -s localhost:8000/.well-known/agent-card.json
curl -s -X POST localhost:8000 -H 'Content-Type: application/json' -H 'A2A-Version: 1.0' -d '{"jsonrpc":"2.0","id":1,"method":"SendMessage","params":{"message":{"messageId":"m1","role":"ROLE_USER","parts":[{"text":"How many words are in: the quick brown fox?"}]}}}'

4. Log in and push

gf agents push takes the agent directory (the one with pyproject.toml), finds the manifest one level in, archives the directory deterministically and uploads it. It prints the archive's digest; the platform stores the same bytes under the same digest, and the CLI refuses a mismatch.

shellCopy
gf login --api "$GENFLEET_API"    # prompts for email and password
gf whoami                         # the account and workspace you are acting as

gf agents digest hello-agent      # optional: the digest, computed offline
gf agents push hello-agent        # ... stored sha256:<64 hex>

Pushing stores the code; it does not create an agent yet. It prints the gf agents create command to run next. You can also do both in one step (next section).

5. Run it in your workspace

First the model key. In the web app, open Secrets and create a secret named OPENROUTER_API_KEY. The agent only receives it because its spec references it (below); a secret nothing references reaches no agent.

Then create the agent from the pushed code, reading that secret:

shellCopy
gf agents create sha256:<the digest from gf agents push> \
  --name hello-agent --secret OPENROUTER_API_KEY

# or push, create and start it in one go:
gf agents push hello-agent --create --secret OPENROUTER_API_KEY --activate

--secret OPENROUTER_API_KEY passes a reference, not your key. The agent's spec gets "env": {"OPENROUTER_API_KEY": "$OPENROUTER_API_KEY"}, and the platform replaces it with the value of your secret of that name each time the agent starts. The key never passes through gf. With --create, the name and description come from genfleet.toml. Use --name to override the name.

6. Chat with it

Start the agent's sandbox (skip this if you passed --activate), then send it a message:

shellCopy
gf agents activate hello-agent
gf agents chat hello-agent "How many words are in: the quick brown fox?"

Or use the web app. Open Agents, find hello-agent and choose Activate. Then open Chat, pick hello-agent and ask it to count some words. Either way, the answer comes from your count_words tool.

Changed the code? Push again and create the agent under a new name (or delete the old one first). To offer it to other workspaces, submit it to the marketplace from the agent's page; a Genfleet reviewer approves it before it is listed.

When something goes wrong

• gf agents push or create answers 403: your role in the workspace is member. Pushing and creating agents need owner, admin or developer. Ask the workspace owner.
• gf agents create prints “saved but not loaded”: the agent row exists, but the engine could not load it. The message says why. Fix the cause, delete the agent in the web app, and create it again.
• The agent will not activate — check that the secret named in spec.env exists, and that your code imports nothing beyond the SDK and the standard library (step 2).
• An error shows an error id or trace id — include it when you ask for help; it is how we find the failure in our logs.

Where to get help

Reply to your invitation email, or contact the person who invited you. SDK bugs and questions: github.com/genfleet/sdk/issues (public, so never paste keys or private code there).

All Exports

Everything exported from genfleet.sdk.

pythonCopy
from genfleet.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
    TokenUsage,     # Model — token counts (input, output, total)
    AuditEvent,     # Model — structured audit event
    # Config TypedDicts
    ModelConfig,    # Provider + credentials
    MemoryConfig,   # Memory config — platform store or Redis
    MCPConfig,      # MCP server connection config
    AuditConfig,    # Audit backend config
    # Tools
    ToolWrapper,    # Class — wraps callable as ToolProtocol
    tool,           # Decorator — function → ToolWrapper
)

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