尿应平图系统计

1. Install

pip install calfkit

<br>

3. Define and Deploy the Tool Node

Define and deploy a tool as an independent service. Tools are not owned by or coupled to any specific agent—once deployed, any agent in your system can discover and invoke the tool. Deploy once, use everywhere.

```python

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

async def main(): client = Client.connect("localhost:9092") # Connect to Kafka broker worker = Worker(client, nodes=[get_weather]) # Initialize a worker with the tool node await worker.run() # (Blocking call) Deploy the service to start serving traffic

if name == "main": asyncio.run(main())

Run the file to deploy the tool service:

<br>

4. Deploy the Agent Node

Deploy the agent as its own service. The Agent handles LLM chat, tool orchestration, and conversation management in a single node. Import the tool definition to register it with the agent—the tool definition is reusable and does not couple the agent to the tool's deployment.

```python

Quick Start

Structured Outputs (Optional)

Agents can be deployed with a final_output_type to enforce structured output from the LLM. The output is type-safe and deserialized 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"

<br>

Client-Side Features (Optional)

The Client supports multi-turn conversations, runtime dependency injection, and temporary instruction overrides—all without redeploying the agent.

Multi-turn conversations — pass the message history from a previous result to maintain context:

```python result = await client.execute_node("What's the weather in Tokyo?", "agent.input")

Gating Node Invocations (Optional)

When multiple agents share an input topic (each with its own consumer group), every agent receives every message published to that topic. A gate stack lets a node decide whether to handle an inbound event before run() runs — avoiding wasted LLM tokens on messages addressed elsewhere.

Gates are predicates: Callable[[SessionRunContext], bool | Awaitable[bool]]. They stack with AND semantics in registration order and short-circuit on the first False, exception, or non-bool return. When any gate rejects, run() is skipped and the envelope is returned unchanged — the Kafka offset still commits.

Constructor form — good for shared, cross-cutting predicates passed in as values:

def is_scheduler_target(ctx) -> bool:
    discord = ctx.deps.provided_deps.get("discord", {})
    return discord.get("slash_target") == "scheduler"

scheduler = Agent(
    "scheduler",
    subscribe_topics="discord.thread.123",
    model_client=OpenAIResponsesModelClient(model_name="gpt-5.4-nano"),
    gates=[is_scheduler_target],
)

Decorator form — good for node-specific gates defined inline:

scheduler = Agent("scheduler", subscribe_topics="discord.thread.123", model_client=...)

@scheduler.gate
def is_scheduler_target(ctx) -> bool:
    discord = ctx.deps.provided_deps.get("discord", {})
    return discord.get("slash_target") == "scheduler"

Constructor and decorator forms can be combined; constructor gates run first.

Idempotency requirement: Kafka may redeliver an event before its offset commits, so gates may run more than once for the same logical message. Keep gate functions deterministic and side-effect-free.

Failure behavior: If a gate raises or returns a non-bool, the framework logs the failure and rejects the message (fail-safe). Place cheap fast-reject gates first to maximize short-circuit efficiency.

For tool-node gating, pass gates=[...] to ToolNodeDef.create_tool_node(...) directly; the @agent_tool decorator doesn't expose gates= because tool topics are typically 1:1.

<br>

Consumer Nodes (Optional)

A consumer node is a terminal sink — it subscribes to one or more topics and runs arbitrary Python logic against every event flowing through. Consumers receive the same NodeResult that Client.execute_node() returns, including the full session state (tool_calls, tool_results, message_history, metadata).

Deploy a consumer as its own service. Wire it to an agent's publish_topic (or any topic carrying calfkit envelopes) to observe outputs from agents, tools, and intermediate hops:

```python

🐮 Calfkit SDK

The SDK to build AI agents as orchestratable, event-driven microservices.

Calfkit lets you compose agents as decoupled microservices–agents, tools, workflows–that communicate asynchronously. Add agents to teams without hardcoding any orchestration logic, scale each component independently, and stream agent outputs to any downstream listener.

pip install calfkit

<br>