KCS Agent

A typed, provider-neutral Python agent loop with streaming output and composable extensions. Python 3.12+ · MIT · No LangChain dependency.

Install

uv add kcs-agent
# or: pip install kcs-agent

Quick start

import asyncio
import os

from kcs_agent import Agent, DeepSeekProvider, ReasoningEffort, UserMessage, UserMessageData


async def main() -> None:
    model = DeepSeekProvider(model="deepseek-v4-flash", api_key=os.environ["DEEPSEEK_API"])
    try:
        result = await Agent(model).run(
            UserMessage(data=UserMessageData("Explain an agent loop in one sentence.")),
            reasoning_effort=ReasoningEffort.OFF,
        )
        print(result.message.data.content)
    finally:
        await model.aclose()


asyncio.run(main())

Each invocation supplies the latest input. The core does not retrieve earlier conversations. It owns the bounded model/tool loop, ordered events, cancellation state, and usage accumulation.

Tools and system-prompt guidance

A tool remains a regular typed function. Its first docstring paragraph is its description. The bare @tool form reads Snippet:, Guidelines:, and Args: from the docstring:

from kcs_agent import Agent, ToolPromptExtension, tool


@tool
def add(left: int, right: int) -> int:
    """Add two integer values.

    Snippet:
        add(left, right) -> sum

    Guidelines:
        - Use for exact integer addition.

    Args:
        left: First integer.
        right: Second integer.

    Examples:
        >>> add(2, 3)
        5
    """
    return left + right


def build_agent(model):
    return Agent(
        model,
        system_prompt="Answer accurately.",
        extensions=[ToolPromptExtension([add])],
    )

The configured form, @tool(snippet="...", guideline="..."), overrides docstring guidance. Examples: documents Python usage; it is never inserted into the model prompt.

ToolPromptExtension registers tools and appends their guidance to the end of the combined system instructions, first all snippets, then all guidelines. It derives each model request from unchanged run messages, preventing duplicate guidance in repeated tool rounds. Place it after other prompt-transforming extensions.

ToolExtension registers tools without modifying the prompt. parse_tool(function) exports the callable's input and return JSON Schemas. Arguments are validated and nested Pydantic models are constructed before invocation. Synchronous tools run in a worker thread. Cancellation does not forcibly terminate an already-running synchronous function; applications must provide cooperative cancellation for long-running side effects.

History extension

HistoryExtension invokes an async loader once per run. The loader receives typed AgentContext[SessionDataT], including the stable session ID. It returns preceding messages, optionally a context snapshot followed by its replay tail. Exclude the current input, even if the application already persisted it.

from kcs_agent import (
    Agent, AgentContext, AnyMessage, HistoryExtension, SessionState,
    ToolPromptExtension, UserMessage, UserMessageData,
)

history: dict[str, list[AnyMessage]] = {}


async def load_history(context: AgentContext[None]) -> list[AnyMessage]:
    return list(history.get(context.state.session.session_id, ()))


async def conversation(model) -> None:
    session = SessionState(data=None)
    agent = Agent(model, extensions=[HistoryExtension(load_history), ToolPromptExtension([add])])
    for text in ("Use add to compute 2+3.", "What was the result?"):
        result = await agent.run(UserMessage(data=UserMessageData(text)), session=session)
        history[session.session_id] = [
            message for message in result.state.run.messages if message.role != "system"
        ]

This example stores history in memory. Production applications own durable storage and compaction through the loader and lifecycle hooks. Different sessions never share mutable run state through the agent instance.

Streaming and hooks

Agent.stream(...) emits ordered AgentEvent values. Relevant event types include:

• TEXT_DELTA, REASONING_DELTA, and TOOL_CALL_DELTA
• MODEL_STARTED and MODEL_COMPLETED
• TOOL_STARTED, TOOL_COMPLETED, and TOOL_FAILED
• RUN_COMPLETED, RUN_FAILED, and cooperative RUN_CANCELLED

Tool deltas are display-only fragments. The final model response contains authoritative complete tool calls. Models may answer without calling any tool.

An extension can implement load_state, on_run_start, context_messages, tools, before_model, after_model, before_tool, after_tool, on_message, on_checkpoint, on_error, on_run_end, and release_state. Hooks execute in registration order. before_model may return a replacement immutable ModelRequest; returning None leaves it unchanged. on_message observes generated assistant/tool messages; the application owns persistence of caller-supplied input.

Task cancellation and closing the stream checkpoint partial state and release resources. Task cancellation propagates asyncio.CancelledError; it cannot yield a terminal event to an already disconnected consumer.

Messages and providers

Messages separate semantic data from typed metadata using Message[DataT, MetadataT]. Concrete roles are system, user, assistant, and tool. User content accepts text and ordered image blocks with URL, bytes, or opaque asset sources. Applications must resolve opaque asset IDs to actual images when visual understanding is required. Ollama accepts encoded image bytes or base64 data URLs; remote images must be downloaded by the application first.

Official SDK adapters are included:

Adapters use the operating system certificate trust store. Each accepts an optional httpx.AsyncBaseTransport for isolated tests and exposes aclose().

Reasoning effort is supplied per run using ReasoningEffort: off, minimal, low, medium, high, and xhigh. Provider capabilities differ. DeepSeek V4 maps minimal/low to low, medium/high to high, and xhigh to max. Anthropic and Google use token budgets. OpenAI forwards enabled levels; unsupported model/level combinations may be rejected by the provider. Ollama maps enabled levels to its boolean thinking switch.

For structured output, ModelRequest.tool_choice names a schema-bound response tool. OpenAI, Anthropic, and Google force that choice; Ollama receives an explicit instruction. Consumers must validate the completed tool arguments against their output model.

Development

uv sync
uv run ruff check src tests examples
uv run ruff format --check src tests examples
uv run pytest

# Real DeepSeek streams and tool round trips via both protocols:
KCS_AGENT_LIVE_TESTS=1 uv run pytest tests/test_live_providers.py

Live tests require DEEPSEEK_API and default to deepseek-v4-flash. Optional overrides: DEEPSEEK_API_MODEL, DEEPSEEK_ANTHROPIC_MODEL, DEEPSEEK_ANTHROPIC_BASE_URL, and DEEPSEEK_ANTHROPIC_API_KEY. Network and provider failures fail explicitly when live tests are enabled.

uv build
uv publish dist/kcs_agent-0.1.0.tar.gz dist/kcs_agent-0.1.0-py3-none-any.whl

Supply publishing credentials through your local credential store or UV_PUBLISH_TOKEN.