acp-agent-framework

---

How It Works

Why ACP?

Features

Install

pip install acp-agent-framework

For development:

git clone https://github.com/sanjay3290/agentic-framework-acp.git
cd agentic-framework-acp
pip install -e ".[dev]"

Quick Start

Simple Agent

from acp_agent_framework import Agent, serve

agent = Agent(
    name="assistant",
    backend="claude",  # or "gemini" or "codex"
    instruction="You are a helpful coding assistant.",
)

serve(agent)  # Serves over ACP stdio

Sequential Pipeline

Chain agents — each one's output feeds into the next:

from acp_agent_framework import Agent, SequentialAgent, serve

researcher = Agent(
    name="researcher",
    backend="claude",
    instruction="Research the given topic thoroughly.",
    output_key="research",
)

summarizer = Agent(
    name="summarizer",
    backend="claude",
    instruction=lambda ctx: f"Summarize:\n\n{ctx.state.get('research', '')}",
)

pipeline = SequentialAgent(name="research-pipeline", agents=[researcher, summarizer])
serve(pipeline)

Router Agent

Route requests to specialists based on keywords:

from acp_agent_framework import Agent, Route, RouterAgent, serve

code_agent = Agent(name="coder", backend="claude", instruction="Help with code.")
writing_agent = Agent(name="writer", backend="gemini", instruction="Help with writing.")

router = RouterAgent(
    name="smart-router",
    routes=[
        Route(keywords=["code", "python", "bug"], agent=code_agent),
        Route(keywords=["write", "essay", "email"], agent=writing_agent),
    ],
    default_agent=code_agent,
)

serve(router)

Custom Tools

Wrap Python functions as tools the LLM can invoke via MCP:

from acp_agent_framework import Agent, FunctionTool, serve

def search_docs(query: str) -> str:
    """Search documentation for a query."""
    return f"Results for: {query}"

agent = Agent(
    name="tool-agent",
    backend="claude",
    instruction="Use tools to help the user.",
    tools=[FunctionTool(search_docs)],
)

serve(agent)

Tools are bridged to the backend via MCP — each FunctionTool is exposed as an MCP tool that the LLM can call during inference.

ToolAgent (No LLM Needed)

For deterministic workflows and data pipelines:

from acp_agent_framework import ToolAgent, FunctionTool, Context

async def build_report(ctx, tools):
    data = tools["fetch_data"].run({"source": "api"})
    return tools["format_report"].run({"data": data})

agent = ToolAgent(
    name="reporter",
    tools=[FunctionTool(fetch_data), FunctionTool(format_report)],
    execute=build_report,
    output_key="report",
)

Guardrails

Validate and transform inputs/outputs:

from acp_agent_framework import Agent, Guardrail, serve

agent = Agent(
    name="safe-agent",
    backend="claude",
    instruction="Be helpful.",
    input_guardrails=[
        Guardrail("pii-redact", lambda t: t.replace("SSN: 123-45-6789", "SSN: [REDACTED]")),
    ],
    output_guardrails=[
        Guardrail("length-limit", lambda t: t[:500] + "..." if len(t) > 500 else t),
    ],
)

Agent Skills

Load reusable capabilities per the agentskills.io spec:

agent = Agent(
    name="chat-agent",
    backend="claude",
    instruction="You are a helpful assistant.",
    skills=["google-chat"],  # loads from .agents/skills/google-chat/SKILL.md
)

Streaming

Real-time token-by-token output:

agent = Agent(name="storyteller", backend="gemini", instruction="Write stories.", stream=True)

async for event in agent.run(ctx):
    if event.type == "stream_chunk":
        print(event.content, end="", flush=True)

Agent-to-Agent Delegation

Wrap any agent as a tool for another agent:

from acp_agent_framework import Agent, AgentTool, serve

translator = Agent(name="translator", backend="gemini", instruction="Translate text.")
summarizer = Agent(name="summarizer", backend="claude", instruction="Summarize text.")

manager = Agent(
    name="content-manager",
    backend="claude",
    instruction="Use translator and summarizer tools to help the user.",
    tools=[AgentTool(translator, cwd="."), AgentTool(summarizer, cwd=".")],
)

serve(manager)

Serving Agents

HTTP Endpoints

CLI

acp-agent init my-agent                            # Scaffold a new project
acp-agent run my_agent.agent:agent                 # Run over ACP stdio
acp-agent run my_agent.agent:agent -t http -p 8000 # Run as HTTP server

Supported Backends

Any other agent from the ACP Registry can be used by registering its launch command.

Register custom backends:

from acp_agent_framework import BackendConfig, BackendRegistry

registry = BackendRegistry()
registry.register("my-backend", BackendConfig(
    command="my-agent-binary",
    args=["--acp"],
))

Architecture

Testing

# Unit tests (174 tests)
pytest tests/ -v

# Integration tests (requires backends installed)
pytest tests/ -v -m integration

# Lint
ruff check src/ tests/

Examples

Check the examples/ directory for complete working agents:

Contributing

See CONTRIBUTING.md for development setup and guidelines.

License

MIT