delegent 0.1.1

Async-first framework for tool-using agents with optional multi-agent delegation.

Delegent – Async‑First Agentic Framework 🚀

Delegent is a lightweight, async‑first framework for building tool‑using agents. It supports:

• Simple single‑step agents (SimpleAgentBuilder) and advanced multi‑agent delegation (AgentBuilder).
• Seamless integration with any LLM client (Ollama, OpenAI, Anthropic, Gemini).
• Structured “thinking” streams, observability hooks and metrics.
• Built‑in support for Nuncio multi‑agent delegation, MCP remote tools, and approval‑required tool calls.
• A clean public API that can be imported directly from the top‑level delegent package.

---

📦 Installation

Install the official package via pip:

pip install delegent

Or using uv:

uv pip install delegent

---

✨ Quick Start – Simple Runtime

import asyncio
from delegent import SimpleAgentBuilder, tool, OllamaClient

@tool(
    name="local.greet",
    description="Say hello",
    args_schema={"type": "object", "properties": {"name": {"type": "string"}}},
    keywords=["greet", "hello", "welcome"],
)
async def greet(args):
    return f"hello, {args['name']}"

async def main():
    llm = OllamaClient(model="llama3.1")
    agent = (
        SimpleAgentBuilder()
        .system("You are a helpful planner.")
        .llm(llm, format="json")
        .tools([greet])
        .build()
    )
    result = await agent.run("greet the user", {"name": "world"})
    print(result)

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

---

🧭 Streaming "Thinking" – Structured Events

from delegent import AgentBuilder, OllamaClient, Event

class PrintHandler:
    async def handle(self, event: Event):
        print(event.type, event.data)

agent = (
    AgentBuilder()
    .system("You are a planner.")
    .llm(OllamaClient(model="llama3.1"), format="json")
    .stream(PrintHandler())
    .build()
)

Available event types include planner.started, tool.call, tool.result, final, and, when delegating, delegate.call, delegate.result, etc. Use stream_mode("thinking") for a user‑friendly summary or stream_mode("both") for raw + thinking.

---

🔧 Advanced Features

Multi‑Agent Delegation (Nuncio)

from delegent import AgentBuilder, OllamaClient

agent = (
    AgentBuilder()
    .system("You are a Trading Agent.")
    .llm(OllamaClient(model="llama3.1"), format="json")
    .multi_agent(hub_uri="ws://localhost:8080/nuncio", token="dev-token", role="TRADING_AGENT")
    .delegate_capabilities(["RESEARCH_AGENT"])  # expose sub‑role(s)
    .build()
)
await agent.connect()

Tool Approval (e.g., Payments)

from delegent import AgentBuilder, tool, ApprovalHandler

@tool(
    name="payments.charge",
    description="Charge a customer",
    args_schema={"type": "object", "properties": {"amount": {"type": "number"}}},
    requires_approval=True,
    approval_prompt="Charge the customer?",
    cost_estimate="$100",
)
async def charge(args):
    return {"charged": True, "amount": args["amount"]}

class MyApproval(ApprovalHandler):
    async def approve(self, tool, args):
        # Hook into your UI / consent flow
        return True

agent = AgentBuilder().approval(MyApproval()).tools([charge]).build()

MCP Remote Tools

from delegent import AgentBuilder, MCPClientConfig, MCPTool

async def call_tool(name, args):
    # Call your remote service here
    return {"ok": True}

config = MCPClientConfig(
    tools=[MCPTool(name="mcp.search", description="Search", args_schema={"type": "object"})],
    call_tool=call_tool,
)
agent = AgentBuilder().mcp(config).llm(OllamaClient(model="llama3.1"), format="json").build()

---

📚 Documentation

• Specification – docs/FRAMEWORK.md
• Full User Guide – docs/USER_GUIDE.md
• Cookbook (Examples) – docs/COOKBOOK.md

All documentation is rendered on the GitHub repository and is also displayed on PyPI via the long_description field.

---

🤝 Contributing

We welcome contributions! Please:

1. Fork the repository.
2. Create a feature branch (git checkout -b my‑feature).
3. Run the test suite locally (uv run python -m pytest).
4. Submit a Pull Request.

For detailed guidelines, see CONTRIBUTING.md.

---

📄 License

Released under the Apache License 2.0. See the bundled LICENSE file for details.

---

📬 Contact & Support

• GitHub Issues: https://github.com/nuncio-labs/delegent/issues
• Email: [e-mail]

---

Enjoy building next‑generation agents with Delegent!

This framework is intentionally simple by default and modular:

• Create agents with a system prompt
• Attach local tools easily
• Mirror MCP tools into the same registry
• Stream structured “thinking” events
• Use any LLM client (Ollama, OpenAI, Anthropic, Gemini)
• Use Nuncio for multi‑agent delegation (with streaming)

Simple runtime uses SimpleAgentBuilder and only supports action + final plans. Advanced features (delegate/approval) use AgentBuilder.

---

Documentation Map

• Specification: docs/FRAMEWORK.md
• Full Guide: docs/USER_GUIDE.md
• Practical Examples (Cookbook): docs/COOKBOOK.md

---

Quick Start (Simple Runtime)

import asyncio
from delegent import SimpleAgentBuilder, tool, OllamaClient


@tool(
    name="local.greet",
    description="Say hello",
    args_schema={"type": "object", "properties": {"name": {"type": "string"}}},
    keywords=["greet", "hello", "welcome"],
)
async def greet(args):
    return f"hello, {args['name']}"


async def main():
    ollama = OllamaClient(model="llama3.1")
    agent = (
        SimpleAgentBuilder()
        .system("You are a helpful planner.")
        .llm(ollama, format="json")
        .tools([greet])
        .build()
    )
    result = await agent.run("greet the user", {"name": "world"})
    print(result)


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

---

Streaming “Thinking” (Structured Events)

from delegent import AgentBuilder, OllamaClient, Event


class PrintHandler:
    async def handle(self, event: Event):
        print(event.type, event.data)


agent = (
    SimpleAgentBuilder()
    .system("You are a planner.")
    .llm(OllamaClient(model="llama3.1"), format="json")
    .stream(PrintHandler())
    .build()
)

Events emitted: planner.started, planner.plan, tool.call, tool.result, final In delegate mode, you also get: peer.dialog.started, peer.dialog.ended, tool.stream, delegate.call, delegate.result

New lifecycle/correlation events include: run.started, planner.call, planner.error All runtime events include run_id (auto-generated if absent). If provided in context, session_id is propagated in event payloads.

Stream Modes

Stream output can be selected with:

• raw (default): existing technical events unchanged.
• thinking: user-friendly thinking summaries.
• both: thinking summaries plus prefixed raw events (raw.<event_type>).

agent = (
    AgentBuilder()
    .system("You are a planner.")
    .llm(OllamaClient(model="llama3.1"), format="json")
    .stream_mode("thinking")
    .build()
)

You can also attach a custom thinking sink:

class MyThinkingHandler:
    async def handle(self, event):
        print("THINK:", event.type, event.data)

agent = AgentBuilder().stream_mode("thinking").thinking_handler(MyThinkingHandler()).build()

Thinking mode currently includes planner thought text and links parent run_id with delegated child run IDs when available. Friendly stream is optimized for readability; raw stream remains the source of truth for debugging.

Delegation is not a tool. The planner should return a delegate plan:

{
  "type": "delegate",
  "thought": "Ask the research agent for a signal.",
  "agent_role": "RESEARCH_AGENT",
  "query": "Provide a BUY/NO_BUY signal for AAPL",
  "context": { "symbol": "AAPL" }
}

---

Preferred Tools (Soft Bias)

The planner prefers tools when relevant. You can also override explicitly:

result = await agent.run(
    "greet the user",
    {"name": "world", "preferred_tools": ["local.greet"]},
)

---

Planning History

The planner receives step history and a summary (when history is long).

agent = AgentBuilder().history_limit(5).build()

Planner context includes:

• history: recent steps (plan + results)
• history_summary: compact summary of older steps

---

Observability (Opt-In)

Metrics

from delegent import AgentBuilder, FrameworkMetricsCollector

collector = FrameworkMetricsCollector()
agent = AgentBuilder().llm(ollama, format="json").metrics(collector).build()

Structured Logging (JSON stdout)

agent = AgentBuilder().llm(ollama, format="json").logging().build()

Or provide config:

agent = AgentBuilder().logging({"level": "INFO", "redact_keys": ["email"]}).build()

Internal Tracing (In-Memory Spans)

from delegent import AgentBuilder, TraceCollector

trace_collector = TraceCollector()
agent = AgentBuilder().llm(ollama, format="json").tracing(trace_collector).build()

No observability handler is auto-enabled by default.

---

Tool Approval (Payments / Sensitive Actions)

Mark tools as requiring approval and provide a handler:

from delegent import AgentBuilder, tool, ApprovalHandler

@tool(
    name="payments.charge",
    description="Charge a customer",
    args_schema={"type": "object", "properties": {"amount": {"type": "number"}}},
    requires_approval=True,
    approval_prompt="Charge the customer?",
    cost_estimate="$100",
)
async def charge(args):
    return {"charged": True, "amount": args["amount"]}

class MyApproval(ApprovalHandler):
    async def approve(self, tool, args):
        return True  # integrate your UI/consent flow here

agent = AgentBuilder().approval(MyApproval()).tools([charge]).build()

---

MCP Tools (Registry‑Backed)

from delegent import AgentBuilder, MCPClientConfig, MCPTool

async def call_tool(name, args):
    # call your MCP server here
    return {"ok": True}

config = MCPClientConfig(
    tools=[MCPTool(name="mcp.search", description="Search", args_schema={"type": "object"})],
    call_tool=call_tool,
)

agent = AgentBuilder().mcp(config).llm(ollama, format="json").build()

---

Nuncio Multi‑Agent (Delegate Mode)

from delegent import AgentBuilder, OllamaClient

agent = (
    AgentBuilder()
    .system("You are a Trading Agent.")
    .llm(OllamaClient(model="llama3.1"), format="json")
    .multi_agent(hub_uri="ws://localhost:8080/nuncio", token="dev-token", role="TRADING_AGENT")
    .delegate_capabilities(["RESEARCH_AGENT"])
    .build()
)

await agent.connect()

---

Memory and Sessions

• Memory APIs support explicit session scoping:
  • add(text, metadata=None, session_id=None)
  • retrieve(query, k=5, session_id=None)
• The agent passes context.session_id explicitly into memory calls.
• Legacy store behavior without explicit session_id remains supported for compatibility.

For parallel calls on the same Agent instance:

• Agent.run is asyncio non-blocking and can run concurrently.
• Session-scoped memory isolation depends on passing distinct session_id values in context.
• Blocking tool handlers can still block the event loop; tools should be async-safe.

---

LLM Clients

Built‑in:

• OllamaClient
• OpenAIClient
• AnthropicClient
• GeminiClient

All implement: complete(prompt, **kwargs)

---

Public API

• SimpleAgentBuilder (simple runtime)

  • .system(prompt)
  • .llm(client, **kwargs)
  • .tools([...])
  • .mcp(config)
  • .stream(handler)
  • .build() -> Agent

• AgentBuilder

  • .system(prompt)
  • .llm(client, **kwargs)
  • .tools([...])
  • .mcp(config)
  • .nuncio(config) (legacy remote tool adapter)
  • .remote_tool(name, role) (legacy)
  • .multi_agent(hub_uri, token, role, capabilities)
  • .delegate_capabilities(roles)
  • .stream(handler)
  • .stream_mode("raw" | "thinking" | "both")
  • .thinking_handler(handler)
  • .metrics(collector)
  • .logging(handler_or_config=None)
  • .tracing(handler_or_collector=None)
  • .allow_delegate(enabled)
  • .delegate_timeout(seconds)
  • .build() -> Agent

• Agent

  • .run(query, context=None)
  • .stream(query, context=None) → async generator of events
  • .connect() / .close() for delegate mode