Harnessed

A composable Python agent SDK. Keep tools, instructions and event hooks together in plugins, combine them into reusable kits, and stream typed events from a live session. Harnessed owns the agent loop and plugin lifetimes.

Python 3.12-3.14, Pydantic v2. Optional provider integrations use PydanticAI's Direct API.

Harnessed 0.x is alpha software. APIs and persisted session formats may change in minor releases before 1.0. See the release notes and compatibility policy. For the breaking prompt changes in 0.10.0, follow the upgrade from 0.9.0.

Build an agent

uv add "harnessed[openai]"

import asyncio

from harnessed import Agent, Instructions, Plugin, PluginOrder, SystemPrompt, tool
from harnessed.events import ToolCallFinishedEvent
from harnessed.kernel import hook


class Calculator(Plugin):
    def instructions(self) -> SystemPrompt:
        return SystemPrompt("Use add for arithmetic instead of calculating mentally.")

    @tool(readonly=True)
    async def add(self, a: int, b: int) -> int:
        """Add two integers."""
        return a + b

    @hook(ToolCallFinishedEvent)
    def on_result(self, event: ToolCallFinishedEvent) -> None:
        print(f"{event.call.name}: {event.result.status}")


calculator = Calculator()
agent = Agent(
    model="deepseek:deepseek-chat",  # DEEPSEEK_API_KEY
    plugins=[calculator],
    instructions="You are a helpful assistant. Keep answers concise.",
    prompt=SystemPrompt.from_plugins(order=PluginOrder.AS_LISTED),
)


async def main() -> None:
    result = await agent.run("What is 19 + 23?")
    print(result.text)


asyncio.run(main())

One plugin colocates the tool, its model-facing instructions and its event hook. @tool derives the input schema from Python types; the method docstring describes the tool. @hook observes the selected event type across the session. SystemPrompt.from_plugins(order=PluginOrder.AS_LISTED) explicitly collects complete instruction blocks in resolved plugin-list order.

Agent is reusable configuration. agent.run() opens a session for one turn; use agent.session() when several turns should share a conversation.

Stream typed events

Using the agent above, handle the events your application needs with ordinary Python class patterns:

from harnessed.events import (
    TextDeltaEvent,
    ToolCallFinishedEvent,
    ToolCallRequestedEvent,
    ToolProgressEvent,
    TurnFinishedEvent,
)


async def conversation() -> None:
    async with agent.session() as session:
        async for event in session.stream("What is 19 + 23?"):
            match event:
                case TextDeltaEvent(text=text):
                    print(text, end="", flush=True)
                case ToolCallRequestedEvent(call=call):
                    print(f"\nCalling {call.name}")
                case ToolProgressEvent(text=text):
                    print(f"\n{text}")
                case ToolCallFinishedEvent(call=call, result=result):
                    print(f"\n{call.name}: {result.status}")
                case TurnFinishedEvent(result=result):
                    print(f"\nFinished: {result.stop_reason}")

        follow_up = await session.run("Add 8 to that result.")
        print(follow_up.text)

Events are immutable typed values with stable tags such as text.delta and tool.call.finished. The stream includes:

ToolCallFinishedEvent carries the final tool result, including denied or interrupted calls. TurnFinishedEvent carries the assembled run result on normal stream completion. Plugins can also hook audit events such as StepEndedEvent and TurnEndedEvent, or declare their own event types. See the event reference for all events and delivery surfaces, and streaming for approval handling and stream cleanup.

Compose capabilities with Kit

Bundle library defaults, nest kits, and replace individual policies at the application boundary:

from harnessed import Agent, Kit, PluginOrder, SystemPrompt
from harnessed.plugins import AgentLoop, Compaction, Filesystem, Shell


def workspace_kit(root: str) -> Kit:
    return Kit(
        "workspace",
        [
            Filesystem(root),
            Shell(root),
            AgentLoop(max_steps=30),
            Compaction(keep_steps=4),
        ],
        prompt=SystemPrompt.from_plugins(order=PluginOrder.AS_LISTED),
    )


workspace = workspace_kit(".").without(Shell)
agent = Agent(
    model="deepseek:deepseek-chat",
    plugins=[workspace, AgentLoop(max_steps=10)],
    instructions="Help the user understand this codebase. Cite file paths in your answer.",
)

print(agent.describe())

without(Shell) removes that capability from the kit. The direct AgentLoop replaces the kit's loop through its loop slot; custom plugins can declare slots with class MyPolicy(Plugin, slot="my.policy"). Unslotted plugins are additive. SystemPrompt.from_plugins(order=PluginOrder.AS_LISTED) follows the selected plugins, and agent.describe() shows their origins, replacements and exclusions before any session starts.

Tools, approvals, output limits, compaction and tracing all compose through plugins. See plugin composition for nested-kit precedence and lifecycle rules.

Inject services into tools

Separate a backend from the capabilities that use it. Service publishes an implementation; Inject[T] supplies it without exposing it as a model argument:

from pathlib import PurePath

from harnessed import Agent, Plugin, ToolContext, tool
from harnessed.kernel import Inject
from harnessed.plugins import Service
from harnessed.plugins.filesystem import FileSystemBackend, LocalFileSystem


class ProjectDocs(Plugin, needs=(FileSystemBackend,)):
    @tool(readonly=True)
    async def read_document(
        self, path: str, *, files: Inject[FileSystemBackend], ctx: ToolContext
    ) -> str:
        """Read up to 4,000 characters from a workspace document."""
        await ctx.progress(f"Reading {path}")
        return await files.read(PurePath(path), limit=4_000)


agent = Agent(
    model="deepseek:deepseek-chat",
    plugins=[Service(FileSystemBackend, LocalFileSystem(".")), ProjectDocs()],
)

The model supplies only path. ToolContext provides per-call progress and cancellation; needs=(FileSystemBackend,) ties the plugin's lifetime to that service. Removing the provider suspends its dependent tools; restoring it creates a fresh activation. Backends implement ordinary Python protocols, so the same tools can use a local, remote or sandboxed implementation.

Add capabilities for part of a session

Using Calculator from the first example, mount its tools, instructions and hooks for a bounded scope:

from harnessed import Agent, SectionDelivery


async def temporary_tools() -> None:
    agent = Agent(model="deepseek:deepseek-chat")
    async with agent.session() as session:
        async with await session.use(Calculator(), section_delivery=SectionDelivery.CONTEXT):
            result = await session.run("What is 19 + 23?")
            print(result.text)

Leaving the inner scope removes the tool, hook and mount-owned guidance. SectionDelivery.CONTEXT delivers late instructions as active context; the session's system prompt stays fixed. Each activation has its own lifetime and can keep mutable data in PluginState.

Sessions also expose named context, steering, pause/resume and immutable SessionState snapshots for application-owned persistence. See sessions and context for those controls.

Read by topic

Start with the documentation index or the SDK reference.

• Models and provider settings
• Tools, hooks and cancellation
• Plugin composition and lifecycle
• MCP server tools over stdio and HTTP
• Docker and Bubblewrap sandboxes
• Instructions and changing context
• Sessions, controls and application-owned state
• Typed events and custom extensions
• Approvals and headless operation
• Offline testing
• OpenTelemetry tracing with Phoenix and Langfuse
• Executable examples
• Development and CI
• Changelog and breaking changes

The standalone harnessed.kernel uses only the standard library. Optional features live under harnessed.plugins; providers live under harnessed.models.

Inspired by DeepSeek Harness and Cordis.

Released under the MIT License.