calfkit 0.9.1

Build AI workflows and agents as fully-distributed and event-driven microservices.

🐮 Calfkit SDK

Build AI agents as decentralized, event-driven microservices.

Agents, tools, and consumers are independently deployable nodes that dynamically route messages to each other and communicate asynchronously. Build and compose AI workflows as a distributed network of interconnected AI agents.

$ pip install calfkit

Status: Alpha (pre-1.0). Calfkit is under active development and the public API can change between versions. Pin a version and check the CHANGELOG before big version bumps.

Table of Contents

• Why Calfkit?
• Install
  • Prerequisites
  • Start a Calfkit broker
• Usage
  • 1. Define and deploy the tool node
  • 2. Deploy the agent node
  • 3. Invoke the agent
  • Next steps
    • Structured outputs
    • Deploying to production
• API
• Documentation
• Roadmap
• Contributing
• Contact
• License

Why Calfkit?

Calfkit shifts the paradigm of agent development from orchestration (centralized control) to choreography (decoupled interactions). Building agent teams with traditional agent frameworks rely on tightly coupled, hard-coded interactions. Calfkit allows you to build agents into an interconnected mesh right out of the box.

• Distributed by default: Every agent, tool, and consumer is its own node in a connected network — so agents and tools are a distributed system you deploy and scale piece by piece, not a monolith.

• Stream-native: Agents communicate on event streams, so consuming live data — market feeds, IoT sensors, user activity — is the default, not a bolted-on integration. Plug into any upstream source and publish to any downstream system: CRMs, warehouses, even other agents.

• Composable without coupling: Build multi-agent teams simply by dropping in agents on shared topics — no extra wiring, no edits to existing code.

Install

$ pip install calfkit

The command-line tools (calfkit run, calfkit topics) live behind an optional extra:

$ pip install "calfkit[cli]"

Prerequisites

• Python 3.10 or later
• Docker installed and running (for running with a local Calfkit broker)
• An LLM provider API key

Start a Calfkit broker

Every Calfkit deployment needs a running broker.

Option A: Local Broker (Requires Docker)

Clone the calfkit-broker repo and start a local broker:

$ git clone https://github.com/calf-ai/calfkit-broker && cd calfkit-broker && make dev-up

Once it's ready, open a new terminal to continue.

Option B: ☁️ Calfkit Cloud (In Beta)

Skip the infrastructure. Calfkit Cloud is a fully-managed broker for Calfkit agents — nothing to self-host, with built-in observability and agent-event tracing. You deploy against a hosted broker endpoint instead of running one locally.

Sign up for access →

Usage

A complete, runnable version of this walkthrough lives in examples/quickstart/.

1. Define and deploy the tool node

A tool is an independent service, not owned by any agent. Define one with @agent_tool; once deployed, any agent can invoke it. Deploy once, use everywhere.

# weather_tool.py
from calfkit.nodes import agent_tool

# Define a tool — the @agent_tool decorator turns any function into a deployable tool node
@agent_tool
def get_weather(location: str) -> str:
    """Get the current weather at a location"""
    return f"It's sunny in {location}"

calfkit run points at a module:attr target and starts the tool for you — no extra wiring required. This requires the [cli] extra installed.

$ calfkit run weather_tool:get_weather

2. Deploy the agent node

Provide the Agent with a reference to the tool using the same tool definition. This agent listens to the weather_agent.input topic.

# agent_service.py
from calfkit.nodes import Agent
from calfkit.providers import OpenAIResponsesModelClient
from weather_tool import get_weather  # Import the tool definition (reusable)

agent = Agent(
    "weather_agent",
    system_prompt="You are a helpful assistant.",
    subscribe_topics="weather_agent.input",
    publish_topic="weather_agent.output",  # Stream every agent output here. Other agents and consumers can listen into this stream
    model_client=OpenAIResponsesModelClient(model_name="gpt-5.4-nano"),
    tools=[get_weather],  # Register tool definitions with the agent
)

Set your OpenAI API key:

$ export OPENAI_API_KEY=sk-...

Start the agent:

$ calfkit run agent_service:agent

3. Invoke the agent

Send a message to the agent.

# invoke.py
import asyncio
from calfkit.client import Client

async def main():
    client = Client.connect("localhost:9092")  # Connect to the broker

    # Send a request and await the response
    result = await client.execute_node(
        "What's the weather in Tokyo?",
        "weather_agent.input",  # The topic the agent is listening to
    )
    print(f"Assistant: {result.output}")

if __name__ == "__main__":
    asyncio.run(main())

Run the file to invoke the agent:

$ python invoke.py

Next steps

You've completed the quickstart — a tool and an agent deployed as separate services and invoked over Kafka. The rest of this section covers common things to do next.

Structured outputs

Set final_output_type to enforce structured output from the LLM — it's deserialized into your type automatically on the client side.

from dataclasses import dataclass
from calfkit.nodes import Agent
from calfkit.providers import OpenAIResponsesModelClient

@dataclass
class WeatherReport:
    location: str
    summary: str

agent = Agent(
    "weather_agent",
    system_prompt="You are a helpful assistant.",
    subscribe_topics="weather_agent.input",
    model_client=OpenAIResponsesModelClient(model_name="gpt-5.4-nano"),
    final_output_type=WeatherReport,  # Enforce structured output
)

When invoking, pass the matching output_type to deserialize the response:

result = await client.execute_node(
    "What's the weather in Tokyo?",
    "weather_agent.input",
    output_type=WeatherReport,
)
print(result.output.location)  # "Tokyo"
print(result.output.summary)   # "It's sunny in Tokyo"

Deploying to production

For production, you can deploy each node with an explicit Worker, keeping startup, scaling, and lifecycle management under your control:

# serve_tool.py — deploy the tool as its own service
import asyncio
from calfkit.client import Client
from calfkit.worker import Worker
from weather_tool import get_weather

async def main():
    client = Client.connect("localhost:9092")  # Connect to Kafka broker
    worker = Worker(client, nodes=[get_weather])  # One service per node
    await worker.run()  # (Blocking) serve until stopped

if __name__ == "__main__":
    asyncio.run(main())

$ python serve_tool.py

API

The full public surface is re-exported from the top-level calfkit package:

from calfkit import (
    Client, InvocationHandle, NodeResult,        # client
    Agent, agent_tool, consumer,                 # node authoring
    NodeDef, ToolNodeDef, ConsumerNodeDef, BaseNodeDef,  # node types
    ConsumerFn, GateFunction,                    # node typing helpers
    ToolContext,                                 # tool-side context
    OpenAIModelClient, OpenAIResponsesModelClient, AnthropicModelClient,  # providers
    Worker, LifecycleContext, ServingContext, ResourceSetupContext,       # worker + lifecycle
    ProvisioningConfig,                          # provisioning (config only)
    DeserializationError, LifecycleConfigError, ToolExecutionError,       # exceptions
)

Key entry points:

Symbol	Purpose
Client.connect(server_urls=None, reply_topic=None, reply_ttl=None, *, provisioning=None, ...)	Connect to the broker. Defaults to $CALF_HOST_URL → localhost.
Client.execute_node(prompt, topic, *, output_type=..., deps=..., message_history=..., timeout=None, ...)	Request/reply: publish and await the NodeResult.
Client.invoke_node(...) / Client.emit_to_node(...)	Async-handle and fire-and-forget variants.
Agent(node_id, *, system_prompt=..., subscribe_topics, publish_topic=None, model_client, tools=None, gates=None, final_output_type=str, model_settings=None, ...)	An agent node.
@agent_tool / @consumer(...)	Decorators that turn a function into a tool node / consumer node.
Worker(client, nodes=None, ...) → run() / start() / stop()	Host one or more nodes against the broker.

Full signatures and behavior live in the source docstrings and the documentation below.

Documentation

In-repo documentation lives under docs/.

How-to guides — goal-oriented walkthroughs:

• How to call nodes from a client — the three invocation patterns (execute_node / invoke_node / emit_to_node), multi-turn conversations, runtime dependency injection (deps), temporary instructions, fire-and-forget, and bounding reply memory with reply_ttl.
• How to tap a topic with a consumer node — terminal sinks that run arbitrary Python against every event on a topic; tap an agent's publish_topic to log, persist, or fan out.
• How to gate node invocations — predicate gate stacks that let a node decline an inbound event before run() runs (e.g. when agents share an input topic).
• How to give agents MCP tools — deploy an MCPToolbox fronting an MCP server and pass it to agents like a tool node; tools are discovered and kept fresh across processes automatically.
• Worker lifecycle & embedding — open long-lived resources at startup and close them on shutdown, publish presence events, and run with run(), the embeddable start()/stop(), or async with worker:.

Reference:

• CLI reference — the calfkit run and calfkit topics commands.
• Topic provisioning — the experimental, opt-in topic-creation helper for dev/CI.

Design & background:

• Design documents — accepted and proposed designs.

Roadmap

See ROADMAP.md for what's under consideration — listing there isn't a commitment.

Contributing

Issues and pull requests are welcome. Please open an issue to discuss substantial changes before sending a PR.

This project uses uv for dependency management. Before raising a PR, make sure the checks pass:

$ make fix     # auto-fix lint + formatting
$ make check   # lint, format, and type checks
$ make test    # run the test suite

PR titles follow Conventional Commits (enforced by CI and used to generate the changelog).

Contact

If you found this project interesting or useful, please consider:

• ⭐ Starring the repository — it helps others discover it!
• 🐛 Reporting issues
• 🔀 Submitting PRs

License

This project is licensed under the Apache License 2.0. See the LICENSE file for details.