genfleet-sdk

The developer SDK for the Genfleet agent marketplace platform.

Declare an agent with Agent() — give it a role, model, tools, memory, or MCP connections — and serve it over A2A. The SDK handles provider selection, tool dispatch, session memory, and streaming automatically.

Install

pip install "genfleet-sdk @ git+https://github.com/genfleet/sdk@dev"                    # core (pydantic only)
pip install "genfleet-sdk[openai,serve] @ git+https://github.com/genfleet/sdk@dev"      # OpenAI + A2A server
pip install "genfleet-sdk[anthropic,serve] @ git+https://github.com/genfleet/sdk@dev"   # Anthropic + A2A server
pip install "genfleet-sdk[all] @ git+https://github.com/genfleet/sdk@dev"               # everything

Installed from GitHub rather than PyPI: the package is not published yet, so the git reference is the supported way to get it. Pin a tag or commit instead of @dev if you need a fixed version.

Requires Python 3.12+.

Agent Skill

Install the genfleet-sdk skill so your coding agent understands the SDK and can help you build agents:

npx skills add genfleet/sdk

Your coding agent will automatically use it when working with Agent(), AgentProtocol, @tool, serve(), and all framework integration patterns.

Quickstart

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)

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?"}]}}}'

Publish on Genfleet (private beta)

Invited to the beta? The step-by-step guide — write an agent with a tool, test it locally, gf agents push it, run it in your workspace and chat with it — is the Publish on Genfleet section of the docs: docs/index.html#marketplace.

Agent Constructor

Agent(
    role:    str,                      # system prompt / persona
    model:   ModelConfig,              # provider + credentials
    tools:   list[Callable] = [],      # plain Python functions — auto-wrapped
    mcps:    list[MCPConfig] = [],     # MCP server connections
    memory:  MemoryConfig | "platform" | None = None,  # episodic memory (see below)
    context: str | None = None,        # extra context appended to system prompt
    audit:   AuditConfig | None = None, # structured audit logging
)

ModelConfig — explicit provider/model-name format

{"model": "openai/gpt-4o-mini",       "api_key": "sk-..."}     # OpenAI
{"model": "anthropic/claude-sonnet-4-6", "api_key": "sk-ant-..."} # Anthropic
{"model": "gemini/gemini-1.5-pro",     "api_key": "..."}        # Gemini
{"model": "openrouter/google/gemini-2.5-flash", "api_key": "sk-or-..."}  # any vendor via OpenRouter
{"model": "openai/deepseek-chat",      "api_key": "sk-...", "base_url": "https://api.deepseek.com"}  # any OpenAI-compatible

Model clients — genfleet.model_clients

An agent's model calls go through a model client. The direct providers above are the default. A package can register its own under the entry-point group genfleet.model_clients: a factory (ModelConfig) -> ModelClient | None that returns None when it doesn't apply in the current environment. Installed factories are tried in name order and the first active one is used (the SDK logs which). An explicit Agent(model_client=...) wins over both, which is how tests use genfleet.sdk.models.FakeModelClient.

With a model client installed, model may be a plain id (no provider prefix) and api_key may be left out. ModelConfig["options"] is passed to the client unchanged; the SDK never reads it. During a turn, genfleet.sdk.turn.current_turn() returns the turn's AgentInput.metadata.

Memory

Memory is episodic: one thread of messages per session_id, loaded before a turn and appended to after it.

• On the platform you need not name a backend. A hosted agent is given GENFLEET_MEMORY_URL / GENFLEET_MEMORY_TOKEN by its sandbox and uses the platform's store, scoped to that agent instance. memory="platform" says so explicitly and fails fast outside a sandbox. agent.memory.search(subject, query) and agent.memory.purge(subject) are available on it.
• Anywhere else pass your own Redis, or nothing for a stateless agent.
• Both sandbox variables travel together. With only GENFLEET_MEMORY_URL set, the agent raises at construction; with only GENFLEET_MEMORY_TOKEN, there is nothing to dial, so it runs stateless and logs a warning.
• A retried turn is appended twice unless the caller names it: turn_id, request_id or message_id in the request metadata is passed to the store as the turn's idempotency key.

With Redis memory

agent = Agent(
    role="You are a helpful assistant.",
    model={"model": "openai/gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]},
    memory={
        "type":       "redis",
        "connection": "redis://localhost:6379",
    },
)

Send the same contextId on each message to persist history across turns (it becomes the agent's session_id):

curl -X POST http://localhost:8000 -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"}]}}}'

With MCP server

agent = Agent(
    role="You are a file assistant.",
    model={"model": "openai/gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]},
    mcps=[{
        "type":    "stdio",
        "command": "npx",
        "args":    ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
    }],
)

serve(...) / create_app(...) options for an agent you host yourself: public_url="https://…/" fixes the URL the agent card advertises (otherwise it follows the request's Host and is not cached), and accept_history=True lets a trusted caller send prior turns in metadata["genfleet.history"] (off by default).

With audit logging

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

Console backend — emits structured JSON to the genfleet.audit logger:

audit={"backend": "console"}

File backend — appends JSONL to a file:

audit={"backend": "file", "file_path": "audit.jsonl"}

Callback backend — calls your function (sync or async) for each event:

async def my_handler(event):
    print(event.event_type, event.data)

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

AuditConfig options:

Audit events emitted:

Token usage (input/output/total tokens) is tracked automatically for OpenAI, Anthropic, and Gemini providers and included in llm.response and invocation.end events.

What's in the SDK

@tool decorator

Plain functions passed to tools= are auto-wrapped. Use @tool when you want an explicit name or description:

from genfleet.sdk import tool, ToolCall

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

result = await add.call(ToolCall(id="1", name="add", arguments={"a": 3, "b": 4}))
# "7.0"

Operators and customers

On a hosted agent the platform tells each turn who is writing, as AgentInput.metadata["genfleet.caller"] (CALLER_METADATA_KEY, from genfleet.sdk.caller): role ("operator" or "customer"), id (the verified sender), channel, private (a one-to-one chat), legacy_tools, and name. name is the sender's own display name and is untrusted; don't base decisions on it. Use caller_of(input.metadata) to read it.

Every tool has an audience: who it is offered to. Mark it on the tool:

@tool(audiences=["operator", "customer"])
def order_status(order_id: str) -> str: ...

@tool(audiences=["customer"])               # customers only: staff shouldn't trigger it
def start_return(order_id: str) -> str: ...

An unmarked tool is ["operator"]. An operator in a private chat gets the tools whose audience includes operator. A customer, or an operator writing in a group, gets those whose audience includes customer. So a ["customer"] tool is hidden from staff in a private chat. A call to a tool that wasn't offered is refused as "not found". For MCP servers, the config's customer_safe_tools: [names] (the 0.17 form) marks tools for both audiences; there is no customer-only form in the config, but the platform's setting can make one by name.

Your marking is the default, and the platform's setting wins. The tenant owner sets each tool's audience in the dashboard. The platform sends it as metadata["genfleet.tool_audiences"] (TOOL_AUDIENCES_METADATA_KEY), keyed by tool name, or by manifest slug for tools from load_tools(). Each tool it returns carries its slug, on a fresh proxy that behaves like the original. That setting replaces your marking, and can widen it as well as narrow it. An entry the SDK can't read is offered to no one. Agent applies this to every tool it holds: its own functions, load_tools() tools, remote tools passed in as tools=, and MCP tools. Tools the engine keeps itself are filtered by the engine. Read the effective rule with may_offer(caller, audiences).

customer_safe=True (0.17) still works and means both audiences, but it is deprecated and removed in 1.0, as are may_use and Caller.full_toolset.

Sensitive tools. Mark a tool whose call needs a workspace owner's approval:

@tool(audiences=["operator", "customer"], sensitive=True)
def issue_refund(order_id: str, amount: float) -> str: ...

When the model calls it, it doesn't run. The model is told the action awaits approval, and the platform shows the owner or admin the exact call to approve. Once approved, a follow-up turn runs that one call, exactly once, and the model tells the user the outcome. The engine signs the approved call with a key only this sandbox knows (GENFLEET_APPROVAL_KEY), so another agent or a direct caller can't forge one. On a call from another agent, a sensitive tool is refused. The owner can mark or unmark any tool in the dashboard, and that setting wins.

An agent the platform did not spawn (a local serve, a test) has no roles: a turn without the caller key gets every tool. In a platform sandbox (GENFLEET_HOSTED set; on older engines, GENFLEET_A2A_TOKEN) a missing caller gets the narrowest set, as a malformed one always does. legacy_tools gets every tool. An MCP tool whose name is already taken by another tool is skipped.

Declare who the agent serves in genfleet.toml:

[agent]
name = "support"
audiences = ["operator", "customer"]   # default when absent: operators only

AgentProtocol — direct implementation

For framework integrations (LangChain, CrewAI) that manage their own LLM calls, implement AgentProtocol directly:

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"Got: {input.message}", done=True)

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

Package extras

Examples

See examples/ for complete runnable agents:

Requirements

• Python 3.12+
• pydantic >= 2.7
• Provider extras: [openai], [anthropic], [gemini]
• [serve] extra: fastapi, uvicorn, sse-starlette
• [memory] extra: redis
• [mcp] extra: mcp

License

Apache 2.0 — see LICENSE for details.