fred-sdk

fred-sdk is the authoring library for Fred agents.
It provides everything needed to define agents, tools, workflows, and multi-agent compositions — with no dependency on any running platform service.

Where fred-sdk fits

fred-core          Pure utilities — model factories, embeddings, logging
     │
fred-sdk           Execution engine + authoring surface  ← this package
     │              ReAct, Graph, Team, Deep agent types
     │              Tool authoring, HITL, MCP references
     │
fred-runtime       Platform adapters + pod factory
                    SQL checkpointer, LLM routing, FastAPI app factory

Agent logic belongs in fred-sdk.
Infrastructure wiring (DB, MCP server, Keycloak, object store) belongs in fred-runtime.
fred-sdk must stay importable on a bare laptop with no services running.

Installation

pip install fred-sdk

Requires Python 3.12.

Agent types

ReAct agent

Tool-calling assistant backed by a ReAct loop. The most common agent type.

from fred_sdk import ReActAgent, tool, ToolContext, ToolOutput

class WeatherAgent(ReActAgent):
    agent_id = "my.weather.agent"
    role = "Weather assistant"
    description = "Answers weather questions using the get_weather tool."
    system_prompt_template = "You are a helpful weather assistant."

    @tool("Get current weather for a city")
    async def get_weather(self, city: str, ctx: ToolContext) -> ToolOutput:
        # call external API here
        return ToolOutput.text(f"It is sunny in {city}.")

Graph agent

Deterministic workflow with typed state. Nodes are Python functions; edges and conditional routes are declared in GraphWorkflow.

from fred_sdk import GraphAgent, GraphWorkflow, typed_node, StepResult
from pydantic import BaseModel

class MyState(BaseModel):
    message: str = ""
    result: str = ""

@typed_node(MyState)
async def process(state: MyState, ctx) -> StepResult:
    return StepResult(state_update={"result": f"processed: {state.message}"})

class MyGraphAgent(GraphAgent):
    agent_id = "my.graph.agent"
    role = "Processing pipeline"
    description = "Runs a deterministic processing workflow."
    state_schema = MyState
    workflow = GraphWorkflow(
        entry="process",
        nodes={"process": process},
    )

Graph workflow primitives available from fred_sdk:

Team agent

Multi-agent composition. A coordinator routes or sequences work across members.

from fred_sdk import TeamAgent, AgentSpec

class SupportRouter(TeamAgent):
    agent_id = "my.support.router"
    role = "Support request router"
    description = "Routes support requests to the right specialist."
    mode = "route"
    coordinator_instructions = "Pick the right specialist based on user intent."
    members = (
        AgentSpec(name="Billing", role="Billing questions", agent_ref="my.billing.agent"),
        AgentSpec(name="Technical", role="Technical issues", agent_ref="my.technical.agent"),
    )

Three modes:

Child agents used as agent_ref targets should set public = False so they are not exposed as top-level models in Open WebUI or other OpenAI-compatible frontends.

Deep agent

Extended ReAct variant with a built-in planning step. Inherits the full ReAct authoring surface; the planning engine is wired by the runtime.

from fred_sdk import DeepAgentDefinition

class MyDeepAgent(DeepAgentDefinition):
    agent_id = "my.deep.agent"
    role = "Deep research assistant"
    description = "Plans and executes multi-step research tasks."
    ...

Tool authoring

Tools are async methods decorated with @tool on a ReActAgent subclass.

from fred_sdk import tool, ToolContext, ToolOutput, ToolInvocationError

@tool("Search internal documents for a query")
async def search_docs(self, query: str, ctx: ToolContext) -> ToolOutput:
    token = ctx.access_token          # bearer token from the request
    user_id = ctx.user_id             # current user
    results = await my_search_api(query, token=token)
    if not results:
        raise ToolInvocationError("No documents found.")
    return ToolOutput.text("\n".join(results))

ToolContext gives the tool access to the runtime context: user_id, team_id, session_id, language, access_token, and invoke_agent() for sub-agent calls.

Human-in-the-loop (HITL)

Pause a graph at a node and wait for user input. Use choice_step for menu-driven flows or emit HumanInputRequest directly for free-text prompts.

from fred_sdk import choice_step, HumanChoiceOption

approve_step = choice_step(
    title="Confirm transfer",
    question="Do you want to proceed with this bank transfer?",
    choices=[
        HumanChoiceOption(id="confirm", label="Yes, confirm"),
        HumanChoiceOption(id="cancel",  label="No, cancel"),
    ],
    routes={"confirm": "execute", "cancel": "abort"},
)

MCP server references

Declare which MCP servers an agent needs. The runtime wires the actual connection.

from fred_sdk import MCPServerRef, MCP_SERVER_KNOWLEDGE_FLOW_CORPUS

class MyRagAgent(ReActAgent):
    agent_id = "my.rag.agent"
    ...
    default_mcp_servers = (MCP_SERVER_KNOWLEDGE_FLOW_CORPUS,)

Built-in MCP server constants:

Built-in tool references

Pre-built platform tools declared by reference (no implementation needed in the agent):

from fred_sdk import TOOL_REF_KNOWLEDGE_SEARCH, TOOL_REF_ARTIFACTS_PUBLISH_TEXT

class MyAgent(ReActAgent):
    declared_tool_refs = (TOOL_REF_KNOWLEDGE_SEARCH, TOOL_REF_ARTIFACTS_PUBLISH_TEXT)

Testing your nodes offline

Graph nodes reach the model and sub-agents through two calls: context.invoke_agent(...) and context.invoke_structured_model(...) (used by structured_model_step). fred_sdk.testing ships FakeGraphNodeContext, a double covering exactly those two, so a node's business logic can be tested without a real model or a real sub-agent.

from typing import cast
from fred_sdk import AgentInvocationResult, GraphNodeContext
from fred_sdk.testing import FakeGraphNodeContext

context = FakeGraphNodeContext(
    agent_result=AgentInvocationResult(
        agent_id="my.specialist.agent", structured={"trust": "high"}
    ),
    structured_results={"intent": "question_cloud_general"},
)
result = await my_node(state, cast(GraphNodeContext, context))

assert context.agent_calls[0]["agent_id"] == "my.specialist.agent"

A call you did not configure raises AssertionError immediately, so an under-specified test fails loudly instead of silently returning None into your node's logic.

Running an agent

fred-sdk defines agents; fred-runtime executes them. A minimal pod:

# main.py
from fred_runtime.app import create_agent_app, load_agent_pod_config
from myapp.registry import REGISTRY

config = load_agent_pod_config()
app = create_agent_app(registry=REGISTRY, config=config)

See fred-runtime on PyPI for the full pod setup guide, and fred-samples for a working reference pod.

Knowledge Bases — beta

fred_sdk.knowledge_base tells Fred where a team's documents come from and how to keep them current: declare an identity and configuration fields, write one async handler, call knowledge_base_main(kb). Fred supplies the destination library, a workload identity and a cadence; the handler owns discovery, replay-safe writes and explicit retractions — Fred never infers a deletion from absence. Install fred-sdk[knowledge-base]. Working declarations live in the fred-samples repository under knowledge-bases/.

A Knowledge Base acts as a workload, so it needs a deployment that authenticates and a confidential client of its own. A pod started without its client secret and realm fails immediately, naming what it lacks. A stack running with authentication off cannot host one — including for local development.

This surface is beta: pin your fred-sdk version, as it may change between beta releases. Known limits today:

• Configuration fields are fixed once instances exist — there is no schema migration. Deleting a synchronized folder deletes its documents.
• A run's fate is visible in the workflow engine only. Counters, summaries and issues a handler returns are not stored or displayed, and Fred offers no run history, manual trigger or worker-health check yet.
• There is no inventory or checkpoint API, so a handler cannot ask Fred what it already holds; a connector that loses its own state cannot reconcile deletions.
• A relative path is the document key, so a rename reads as a delete plus an add.
• Cadences are hourly, daily or weekly, with no immediate first run.

The label applies to Knowledge Bases only, not to this SDK's agent APIs.

Related packages

License

Apache License 2.0.