PyBotchi

A deterministic, intent-based AI agent orchestrator with no restrictions. Supports any framework and prioritizes a human-reasoning approach.

---

Core Philosophy

Humans should handle the reasoning. AI should detect intent and translate natural language into processable data.

Traditional development has successfully solved complex problems across every industry using deterministic code, APIs, and events. The real limitation isn't in execution, it's in translation. What if we could accept natural language and automatically route to the right logic?

PyBotchi takes a different approach from most AI frameworks: LLMs excel at understanding intent and translating between human and computer language, not at business logic, calculations, or deterministic execution. Let each do what it does best.

The PyBotchi Workflow

1. Detect & Translate (LLM Layer) - Process natural language to extract intents and identify appropriate Actions with arguments
2. Execute Logic (Your Code) - Traditional code handles business logic, calculations, and data processing
3. Generate Response (LLM Layer) - Transform processed results back into natural language

---

Core Architecture

Nested Intent-Based Supervisor Agent Architecture built on just 3 core classes:

• Action - The central agent with a defined lifecycle for intent and execution logic
• Context - Universal container for conversational state, metadata, and execution context
• LLM - Singleton client for managing your model connection

This minimal foundation ensures extreme speed, efficiency, and maximum customizability.

---

Key Features

Ultra-Lightweight

Only 3 core classes to master. The entire system is built on a minimal foundation that minimizes overhead while maximizing performance.

Object-Oriented Design

Built on Pydantic BaseModel for rigorous data validation and industry-standard type hinting. Every component is inherently overridable and extendable.

JSON Schema Native

Automatic JSON Schema conformance for OpenAI, Gemini, and other LLM providers. Easily adaptable to any provider's specification.

Action Lifecycle Hooks

Fine-grained control over execution stages with overridable hooks: pre, post, on_error, fallback, child_selection, and commit_context.

Highly Scalable

Async-first architecture with built-in support for distributed execution via gRPC. Deploy agents remotely or across machines for massive parallel workloads.

Truly Modular

Agents are isolated, self-contained units. Different teams can independently develop, improve, or modify specific agents without impacting core logic.

Graph By Design

Structured parent-child relationships provide clear visibility into system execution and state, simplifying debugging and testing.

Framework & Model Agnostic

Works with any LLM client, third-party framework, or business requirement. True agnosticism through complete overridability.

MCP Protocol Support

Full integration with Model Context Protocol. Expose your Actions as MCP tools or consume external MCP servers within your workflows.

---

Quick Start

Installation

PyBotchi requires Python 3.12 or higher.

pip install pybotchi

# With gRPC support for distributed execution
pip install pybotchi[grpc]

# With MCP support for Model Context Protocol
pip install pybotchi[mcp]

# With both
pip install pybotchi[grpc,mcp]

Setup LLM

from langchain_openai import ChatOpenAI
from pybotchi import LLM

LLM.add(base=ChatOpenAI(
    api_key="your-api-key",
    model="gpt-4",
    temperature=0.7,
))

Simple Agents

from pybotchi import Action

from pydantic import Field

class Translation(Action):
    """Translate to specific language."""

    message: str = Field(description="The text content to be translated.")
    language: str = Field(description="The ISO code or name of the target language.")

    async def pre(self, context):
        message = await context.llm.ainvoke(f"Reply only with translation of `{self.message}` to {self.language}.")
        await context.add_response(self, message.text)

class MathProblem(Action):
    """Solve math problems."""

    answer: str = Field(description="The answer to the math problem")

    async def pre(self, context):
        await context.add_response(self, self.answer)

Multi-Agent Declaration

from pybotchi import Action

class MultiAgent(Action):
    """AI Assistant for solving math problems and translation."""

    class SolveMath(MathProblem):
        pass

    class Translate(Translation):
        pass

Execution

import asyncio

from pybotchi import Context

async def test():
    context = Context(
        prompts=[
            {"role": "system", "content": "You're an AI that can solve math problems and translate requests."},
            {"role": "user", "content": "4 x 4 then explain in Filipino"},
        ],
    )
    await context.start(MultiAgent)
    print(f"MathProblem: {context.prompts[-3]['content']}")
    print(f"Translate: {context.prompts[-1]['content']}")

asyncio.run(test())

Result:

MathProblem: Four multiplied by four is sixteen. Imagine you have four groups, and each group has four candies. If you count all the candies together, you will have sixteen candies. That's what 4 x 4 means!
Translate: Ang apat na pinarami sa apat ay labing-anim. Ipagpalagay mong may apat na grupo, at bawat grupo ay may apat na kendi. Kung bibilangin mo lahat ng kendi, magkakaroon ka ng labing-anim na kendi. Iyan ang ibig sabihin ng 4 x 4!

Visualize Your Graph

import asyncio

from pybotchi import graph

async def print_mermaid_graph():
    multi_agent_graph = await graph(MultiAgent)
    print(multi_agent_graph.flowchart())

asyncio.run(print_mermaid_graph())

Result:

flowchart TD
__main__.MultiAgent.SolveMath[SolveMath]
__main__.MultiAgent{MultiAgent}
__main__.MultiAgent.Translate[Translate]
__main__.MultiAgent --> __main__.MultiAgent.SolveMath
__main__.MultiAgent --> __main__.MultiAgent.Translate
style __main__.MultiAgent fill:#4CAF50,color:#000000

---

Action Lifecycle

Every Action follows a structured lifecycle that gives you complete control over execution flow:

Flow Control

Lifecycle hooks return ActionResult (ActionReturn | None). The value controls what happens next:

• None (or omit return): continue normally, nothing is interrupted
• ActionReturn.END: stop only this action's remaining lifecycle; siblings at the same level continue unaffected
• ActionReturn.BREAK: stop this action and break the nearest ancestor's __max_iteration__ loop; also stops subsequent siblings
• ActionReturn.STOP: stop the entire agent immediately, propagating through all ancestors
• ActionReturn.stop(value=...): same as STOP but carries a return value accessible after context.start()

Core Lifecycle Hooks

pre - Pre-Execution

Executes before child agents run. Use for:

• Guardrails and validation
• Data gathering (RAG, knowledge graphs)
• Business logic and preprocessing
• Tool execution

child_selection - Agent Selection

Determines which child agents to execute. Override with:

• Traditional control flow (if/else, switch/case)
• Custom LLM routing logic
• Dynamic agent selection

post - Post-Processing

Executes after all child agents complete. Use for:

• Result consolidation
• Data persistence
• Cleanup and recording
• Logging and notifications

on_child_init_error - Child Initialization Error Handling

Handle errors when a child action fails to initialize (e.g. Pydantic validation on LLM tool arguments):

• Feed the error back to the LLM so it can retry on the next iteration
• Record the failed tool call in action history
• Return None to skip the failed action and continue the loop

on_error - Error Handling

Handle errors during execution with:

• Retry mechanisms
• Custom error handling
• Logging and alerts
• Re-raise for parent handling

fallback - Non-Tool Results

Executes when no child agent is selected:

• Process text content results
• Handle non-tool-call responses
• Default behaviors

commit_context - Context Control

Controls context merging with main execution:

• Selective data propagation
• Isolated execution contexts
• Custom synchronization rules

on_max_iteration - Max Iteration Handling

Executes when the __max_iteration__ limit is reached:

• Finalize and return the best available response
• Raise max-iteration-related errors
• Trigger fallback or summarization logic
• Custom additional processing

Extended Lifecycle Hooks

• pre_mcp - MCP connection setup (authentication, config)
• pre_grpc - gRPC connection setup (credentials, metadata)

---

Everything is Overridable & Extendable

class CustomAgent(MultiAgent):
    SolveMath = None  # Remove action

    class NewAction(Action):  # Add new action
        pass

    class Translate(Translation):  # Override existing
        async def pre(self, context):
            # Custom translation logic
            pass

---

Execution Patterns

Sequential Execution

Multiple agents execute in order via iteration or multi-tool calls.

Concurrent Execution

Parallel execution using async patterns. Set __concurrent__ = True on the child actions to run them in parallel:

class ParallelAgent(Action):
    class Task1(Action):
        __concurrent__ = True

    class Task2(Action):
        __concurrent__ = True

Nested Architectures

Build complex hierarchical structures:

class ComplexAgent(Action):
    class StoryTelling(Action):
        class HorrorStory(Action):
            pass
        class ComedyStory(Action):
            pass

    class JokeTelling(Action):
        pass

---

Distributed Systems with gRPC

Scale your agents across multiple servers with real-time context synchronization:

server.py

from pybotchi import Action

from pydantic import Field

class MathProblem(Action):
    """Solve math problems."""

    __groups__ = {"grpc": {"group-1"}}

    answer: str = Field(description="The answer to the math problem")

    async def pre(self, context):
        await context.add_response(self, self.answer)

client.py

from asyncio import run

from pybotchi.grpc import GRPCAction, GRPCConnection, graph


class MultiAgent(GRPCAction):
    __grpc_connections__ = [GRPCConnection("remote", "localhost:50051", ["group-1"])]

    async def pre_grpc(self, context):
        print("Executed pre grpc connection!")


async def print_mermaid_graph():
    multi_agent_graph = await graph(MultiAgent, integrations={"remote": {}})
    print(multi_agent_graph.flowchart())


run(print_mermaid_graph())

Key Benefits

• Unified Graph Execution - Remote Actions integrate seamlessly
• Zero-Overhead Synchronization - No polling loops or coordination overhead
• Database-Free Architecture - Context syncs directly through gRPC
• Concurrent Remote Execution - True distributed parallel processing
• Resource Isolation - Separate compute resources per Action group

Start gRPC server:

pybotchi-grpc server.py

Result

#-------------------------------------------------------#
# Agent ID: agent_8b3c5685c84b4602966d1b3252916aa7
# Agent Path: server.py
# Starting None worker(s) on 0.0.0.0:50051
#-------------------------------------------------------#
# Agent Process: Process-1 [173012]
# Agent Handler: PyBotchiGRPC
#-------------------------------------------------------#

gRPC client print graph:

python3 client.py

flowchart TD
grpc.agent_8b3c5685c84b4602966d1b3252916aa7.MathProblem[MathProblem]
__main__.MultiAgent[MultiAgent]
__main__.MultiAgent --"`**GRPC** : remote`"--> grpc.agent_8b3c5685c84b4602966d1b3252916aa7.MathProblem
style __main__.MultiAgent fill:#4CAF50,color:#000000

---

Model Context Protocol (MCP)

Integrate with the MCP ecosystem. Expose Actions as MCP tools or consume external MCP servers:

As MCP Server

server.py

from pybotchi import Action
from pybotchi.mcp import build_mcp_app

from pydantic import Field


class MathProblem(Action):
    __groups__ = {"mcp": {"group-1"}}

    answer: str = Field(description="The answer to the math problem")

    async def pre(self, context):
        await context.add_response(self, self.answer)


app = build_mcp_app(transport="streamable-http")

As MCP Client

client.py

from asyncio import run

from pybotchi.mcp import MCPAction, MCPConnection, graph


class MultiAgent(MCPAction):
    __mcp_connections__ = [MCPConnection("remote", "SHTTP", "http://localhost:8000/group-1/mcp")]
    # __mcp_connections__ = [MCPConnection("jira", "SSE", "https://mcp.atlassian.com/v1/sse")]

    async def pre_mcp(self, context):
        print("Executed pre mcp connection!")


async def print_mermaid_graph():
    multi_agent_graph = await graph(MultiAgent, integrations={"remote": {}})
    print(multi_agent_graph.flowchart())


run(print_mermaid_graph())

Key Benefits

• Standard Protocol Support - Full MCP specification compatibility
• Group-Based Organization - Fine-grained access control per endpoint
• Bidirectional Integration - Serve or consume MCP tools
• Transport Flexibility - SSE and Streamable HTTP support

Start MCP server:

uvicorn server:app

Result

INFO:     Started server process [6330]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

MCP client print graph:

python3 client.py

flowchart TD
__main__.MultiAgent[MultiAgent]
mcp.remote.MathProblem[MathProblem]
__main__.MultiAgent --"`**MCP** : remote`"--> mcp.remote.MathProblem
style __main__.MultiAgent fill:#4CAF50,color:#000000

---

Examples & Use Cases

Explore practical examples demonstrating PyBotchi's capabilities:

Getting Started

• tiny.py - Minimal implementation
• full_spec.py - Complete feature demonstration

Flow Control

• sequential.py - Sequential action execution
• nested_combination.py - Complex nested structures

Concurrency

• concurrent_combination.py - Async parallel execution
• concurrent_threading_combination.py - Multi-threaded processing

Distributed Systems

• grpc/grpc_pybotchi_agent.py - gRPC server setup
• grpc/grpc_pybotchi_client.py - Distributed orchestration

MCP Integration

• mcp/mcp_pybotchi_agent.py - MCP server implementation
• mcp/mcp_pybotchi_client.py - MCP client integration
• mcp/mcp_pybotchi_client_for_mcp_atlassian.py - Atlassian MCP integration

Real-World Applications

• interactive_action.py - Real-time WebSocket communication

Framework Comparison

• vs/pybotchi_approach.py - PyBotchi implementation
• vs/langgraph_approach.py - LangGraph comparison

---

Why Choose PyBotchi?

Maximum flexibility, zero lock-in. Build agents that combine human intelligence with AI precision.

Perfect for teams that need:

• Modular, maintainable agent architectures
• Framework flexibility and migration capabilities
• Community-driven agent development
• Enterprise-grade customization and control
• Real-time interactive agent communication
• Distributed execution without complexity
• Standard protocol integration (MCP)

---

Known Limitations

Anthropic Model Compatibility

PyBotchi passes only the tools relevant to the current intent during child selection, reducing LLM noise. Anthropic models have a constraint that conflicts with this: once a tool_use / tool_result pair for a given tool appears in the conversation history, Anthropic requires that tool's schema to be included in every subsequent API call — even if the tool is no longer relevant. This can cause validation errors in multi-level agent graphs.

OpenAI does not have this restriction.

Workarounds (choose one):

• 1-level deep agent — Keep all child actions under a single parent so every call includes all schemas.
• add_message instead of add_response — Write plain assistant messages to avoid tool-call entries in history.
• Custom child_selection — Override to re-inject previously-used tool schemas alongside the active ones.

See the full documentation for code examples of each approach.

---

Documentation

Visit our full documentation for:

• Detailed lifecycle hook explanations
• Advanced patterns and best practices
• gRPC and MCP integration guides
• Complete API reference

---

Contributing

We welcome contributions! Whether it's:

• Bug reports and feature requests
• Documentation improvements
• Code contributions
• Example applications

Check out our contributing guidelines to get started.

---

License

PyBotchi is released under the Apache License 2.0. See LICENSE for details.

---

Community

• GitHub: amadolid/pybotchi
• Issues: Report bugs or request features
• Discussions: Join the conversation

---

Ready to build smarter agents? Start with the examples and join the community building the future of human-AI collaboration.