Skip to content

MCP (Model Context Protocol)

MCP integration for AI-Parrot.

MCPToolAdapter

MCPToolAdapter(tool: AbstractTool)

Adapts AI-Parrot AbstractTool to MCP tool format.

Tools marked routing_meta["requires_confirmation"] (e.g. the destructive members of a toolkit's confirming_tools) get an MCP-side guard: a required confirm boolean is injected into their input schema and the call is rejected unless confirm=true is passed — the stdio transport has no interactive HITL channel, so the explicit argument is the confirmation record.

Source code in packages/ai-parrot/src/parrot/mcp/adapter.py
def __init__(self, tool: AbstractTool):
    self.tool = tool
    self.logger = logging.getLogger(f"MCPToolAdapter.{tool.name}")

to_mcp_tool_definition

to_mcp_tool_definition() -> dict[str, Any]

Convert AbstractTool to MCP tool definition.

Source code in packages/ai-parrot/src/parrot/mcp/adapter.py
def to_mcp_tool_definition(self) -> dict[str, Any]:
    """Convert AbstractTool to MCP tool definition."""
    # Extract schema from the tool's args_schema
    input_schema = {}
    if hasattr(self.tool, 'args_schema') and self.tool.args_schema:
        try:
            # Get the JSON schema from the Pydantic model
            input_schema = self.tool.args_schema.model_json_schema()
        except Exception as e:  # noqa: BLE001
            self.logger.warning("Could not extract schema for %s: %s", self.tool.name, e)
            input_schema = {"type": "object", "properties": {}}

    if self._requires_confirmation():
        input_schema.setdefault("type", "object")
        input_schema.setdefault("properties", {})["confirm"] = {
            "type": "boolean",
            "description": (
                "This operation is destructive. Set true ONLY after the "
                "user has explicitly approved it; the call is rejected "
                "otherwise."
            ),
        }
        required = input_schema.setdefault("required", [])
        if "confirm" not in required:
            required.append("confirm")

    return {
        "name": self.tool.name or "unknown_tool",
        "description": self.tool.description or f"Tool: {self.tool.name}",
        "inputSchema": input_schema
    }

execute async

execute(arguments: dict[str, Any]) -> dict[str, Any]

Execute the AI-Parrot tool and convert result to MCP format.

Source code in packages/ai-parrot/src/parrot/mcp/adapter.py
async def execute(self, arguments: dict[str, Any]) -> dict[str, Any]:
    """Execute the AI-Parrot tool and convert result to MCP format."""
    # The guard argument never reaches the tool itself.
    confirm = arguments.pop("confirm", None)
    if self._requires_confirmation() and confirm is not True:
        return {
            "content": [
                {
                    "type": "text",
                    "text": (
                        f"Error: '{self.tool.name}' is a destructive "
                        "operation and was not confirmed. Ask the user "
                        "for approval, then re-invoke with confirm=true."
                    ),
                }
            ],
            "isError": True,
        }
    try:
        # Execute the tool
        result = await self.tool._execute(**arguments)

        # Convert ToolResult to MCP response format
        if isinstance(result, ToolResult):
            return self._toolresult_to_mcp(result)
        else:
            # Handle direct results (for backward compatibility)
            return {
                "content": [
                    {
                        "type": "text",
                        "text": str(result)
                    }
                ],
                "isError": False
            }

    except Exception as e:  # noqa: BLE001
        self.logger.error("Tool execution failed: %s", e)
        return {
            "content": [
                {
                    "type": "text",
                    "text": f"Error executing tool: {e!s}"
                }
            ],
            "isError": True
        }

MCPResource dataclass

MCPResource(uri: str, name: str, description: str | None = None, mime_type: str | None = None)

Represents an MCP Resource.

Resources are read-only data sources exposed by the server.

to_dict

to_dict() -> dict[str, Any]

Convert to MCP protocol dictionary.

Source code in packages/ai-parrot/src/parrot/mcp/resources.py
def to_dict(self) -> dict[str, Any]:
    """Convert to MCP protocol dictionary."""
    data = {
        "uri": self.uri,
        "name": self.name,
    }
    if self.description:
        data["description"] = self.description
    if self.mime_type:
        data["mimeType"] = self.mime_type
    return data

MCPServerBase

MCPServerBase(config: LocalServerConfig)

Bases: ABC

Base class for MCP servers (core, transport-agnostic).

Source code in packages/ai-parrot/src/parrot/mcp/server_base.py
def __init__(self, config: LocalServerConfig):
    self.config = config
    self.tools: dict[str, MCPToolAdapter] = {}

    self.logger = logging.getLogger(f"MCPServer.{config.name}")
    log_level = getattr(logging, config.log_level.upper(), logging.WARNING)
    self.logger.setLevel(log_level)

register_tool

register_tool(tool: AbstractTool)

Register an AI-Parrot tool with the MCP server.

Source code in packages/ai-parrot/src/parrot/mcp/server_base.py
def register_tool(self, tool: AbstractTool):
    """Register an AI-Parrot tool with the MCP server."""
    tool_name = tool.name
    adapter = MCPToolAdapter(tool)
    self.tools[tool_name] = adapter
    self.logger.info("Registered tool: %s", tool_name)

register_tools

register_tools(tools: list[AbstractTool])

Register multiple tools.

Source code in packages/ai-parrot/src/parrot/mcp/server_base.py
def register_tools(self, tools: list[AbstractTool]):
    """Register multiple tools."""
    for tool in tools:
        self.register_tool(tool)

handle_initialize async

handle_initialize(params: dict[str, Any]) -> dict[str, Any]

Handle MCP initialize request.

Source code in packages/ai-parrot/src/parrot/mcp/server_base.py
async def handle_initialize(self, params: dict[str, Any]) -> dict[str, Any]:
    """Handle MCP initialize request."""
    self.logger.info("Initializing MCP server...")

    return {
        "protocolVersion": negotiate_protocol_version(
            params.get("protocolVersion")
        ),
        "capabilities": {
            "tools": {
                "listChanged": False
            }
        },
        "serverInfo": {
            "name": self.config.name,
            "version": self.config.version,
            "description": self.config.description
        }
    }

handle_tools_list async

handle_tools_list(params: dict[str, Any]) -> dict[str, Any]

Handle tools/list request.

Source code in packages/ai-parrot/src/parrot/mcp/server_base.py
async def handle_tools_list(self, params: dict[str, Any]) -> dict[str, Any]:
    """Handle tools/list request."""
    self.logger.info("Listing %s available tools", len(self.tools))

    tools = []
    tools.extend(
        adapter.to_mcp_tool_definition() for adapter in self.tools.values()
    )

    return {"tools": tools}

handle_tools_call async

handle_tools_call(params: dict[str, Any]) -> dict[str, Any]

Handle tools/call request.

Source code in packages/ai-parrot/src/parrot/mcp/server_base.py
async def handle_tools_call(self, params: dict[str, Any]) -> dict[str, Any]:
    """Handle tools/call request."""
    tool_name = params.get("name")
    arguments = params.get("arguments", {})

    self.logger.info("Calling tool: %s with args: %s", tool_name, arguments)

    if tool_name not in self.tools:
        raise RuntimeError(
            f"Tool not found: {tool_name}"
        )

    adapter = self.tools[tool_name]
    return await adapter.execute(arguments)

start abstractmethod async

start()

Start the MCP server.

Source code in packages/ai-parrot/src/parrot/mcp/server_base.py
@abstractmethod
async def start(self):
    """Start the MCP server."""

stop abstractmethod async

stop()

Stop the MCP server.

Source code in packages/ai-parrot/src/parrot/mcp/server_base.py
@abstractmethod
async def stop(self):
    """Stop the MCP server."""

LocalServerConfig dataclass

LocalServerConfig(name: str = 'parrot-mcp-local', version: str = '1.0.0', description: str = '', log_level: str = 'WARNING')

Lightweight config for local-only MCP servers.

LocalMCPServerBase

LocalMCPServerBase(config: LocalServerConfig)

Bases: MCPServerBase

Extension point for local (in-process) MCP transports.

Local transports (e.g. stdio) may reserve stdout as the JSON-RPC channel, so all logging must go to stderr instead of a default handler that could write to stdout.

Source code in packages/ai-parrot/src/parrot/mcp/local_server.py
def __init__(self, config: LocalServerConfig):
    super().__init__(config)
    handler = logging.StreamHandler(sys.stderr)
    handler.setFormatter(
        logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s")
    )
    self.logger.addHandler(handler)
    self.logger.propagate = False

StdioMCPServer

StdioMCPServer(config: LocalServerConfig)

Bases: LocalMCPServerBase

MCP server using stdio transport (core, local-only).

Source code in packages/ai-parrot/src/parrot/mcp/local_server.py
def __init__(self, config: LocalServerConfig):
    super().__init__(config)
    self._request_id = 0
    self._running = False

start async

start()

Start the stdio MCP server.

Source code in packages/ai-parrot/src/parrot/mcp/local_server.py
async def start(self):
    """Start the stdio MCP server."""
    self.logger.info("Starting stdio MCP server with %s tools...", len(self.tools))
    self._running = True
    loop = asyncio.get_running_loop()

    while self._running:
        try:
            # sys.stdin.readline() is blocking — run it off the event loop.
            line = await loop.run_in_executor(None, sys.stdin.readline)
            if not line:
                break

            line = line.strip()
            if not line:
                continue

            try:
                request = json.loads(line)
                response = await self._handle_request(request)

                if response:
                    print(json.dumps(response), flush=True)

            except json.JSONDecodeError as e:
                self.logger.warning("Invalid JSON received: %s", e)
                continue

        except KeyboardInterrupt:
            break
        except Exception as e:  # noqa: BLE001
            self.logger.error("Error in main loop: %s", e)
            continue

    self.logger.info("Stdio MCP server stopped")

stop async

stop()

Stop the stdio server.

Source code in packages/ai-parrot/src/parrot/mcp/local_server.py
async def stop(self):
    """Stop the stdio server."""
    self._running = False

AgentMethodTool

AgentMethodTool(agent: Any, method_name: str, declaration: MCPToolDeclaration)

Bases: AbstractTool

An agent method reified as a real AbstractTool.

Built by :func:build_exposure_set from a @mcp_tool-marked method. name, description and args_schema come from the method's MCPToolDeclaration; _execute() invokes the bound method on the owning agent. The agent is held by weak reference only so this tool never drags the agent into tool-serialization paths and never creates a reference cycle (spec §7 Risks).

The agent is resolved per call, never cached as a bound method, so a reloaded agent (BotManager.reload_agent()) is picked up transparently by any mount holding this tool (spec OQ5).

Initialize the reified tool.

PARAMETER DESCRIPTION
agent

The owning agent instance. Held by weak reference only.

TYPE: Any

method_name

Name of the async method on agent to invoke.

TYPE: str

declaration

The MCPToolDeclaration attached by @mcp_tool.

TYPE: MCPToolDeclaration

Source code in packages/ai-parrot/src/parrot/mcp/agent_tools.py
def __init__(self, agent: Any, method_name: str, declaration: MCPToolDeclaration) -> None:
    """Initialize the reified tool.

    Args:
        agent: The owning agent instance. Held by weak reference only.
        method_name: Name of the async method on `agent` to invoke.
        declaration: The `MCPToolDeclaration` attached by `@mcp_tool`.
    """
    self._agent_ref: weakref.ReferenceType[Any] = weakref.ref(agent)
    self._method_name = method_name
    self._declaration = declaration
    super().__init__(
        name=declaration.name,
        description=declaration.description,
        routing_meta={
            "requires_confirmation": declaration.requires_confirmation,
            "read_only_hint": declaration.read_only_hint,
            "idempotent_hint": declaration.idempotent_hint,
        },
    )
    self.args_schema = declaration.args_schema

MCPToolDeclaration

Bases: BaseModel

Declaration metadata attached by @mcp_tool.

All fields except the hint/limit flags are mandatory — there is no schema inference in v1 (spec §1 Non-Goals). Registration must fail loudly if any mandatory field is missing.

ATTRIBUTE DESCRIPTION
name

MCP tool name exposed to clients.

TYPE: str

description

Human-readable tool description surfaced to MCP clients.

TYPE: str

args_schema

Pydantic model describing the tool's call arguments.

TYPE: type[BaseModel]

returns

Pydantic model describing the tool's return payload.

TYPE: type[BaseModel]

scope

PBAC action/resource scope enforced when the tool is invoked.

TYPE: str

read_only_hint

Maps to the MCP readOnlyHint annotation.

TYPE: bool

idempotent_hint

Maps to the MCP idempotentHint annotation.

TYPE: bool

requires_confirmation

Maps to routing_meta["requires_confirmation"], which MCPToolAdapter already honors (destructiveHint precedent).

TYPE: bool

max_result_tokens

Per-tool result-size cap; None falls back to the mount default.

TYPE: int | None

build_exposure_set

build_exposure_set(agent: Any) -> list[AgentMethodTool]

Scan agent for @mcp_tool-marked methods and build its exposure set.

Walks the agent's class for coroutine methods carrying an MCPToolDeclaration (attached via MCP_TOOL_ATTR), validates there are no name collisions — neither among the decorated methods themselves nor against tools already registered in agent.tool_manager — and reifies each into an AgentMethodTool.

The returned exposure set is a plain list. It is never registered into agent.tool_manager (OQ2) — callers (the MCP mount, TASK-2602) are responsible for what they do with it.

PARAMETER DESCRIPTION
agent

The agent instance to scan. Must expose tool_manager if it has any @mcp_tool-decorated method (used for the collision check); agents with none are never asked for it.

TYPE: Any

RETURNS DESCRIPTION
list[AgentMethodTool]

The agent's exposure set — one AgentMethodTool per decorated

list[AgentMethodTool]

method. Empty if the agent declares none.

RAISES DESCRIPTION
ValueError

If two decorated methods declare the same MCP tool name, or a decorated name collides with an existing agent.tool_manager tool. The message names the agent class and the offending method(s).

Source code in packages/ai-parrot/src/parrot/mcp/agent_tools.py
def build_exposure_set(agent: Any) -> list[AgentMethodTool]:
    """Scan `agent` for `@mcp_tool`-marked methods and build its exposure set.

    Walks the agent's class for coroutine methods carrying an
    `MCPToolDeclaration` (attached via `MCP_TOOL_ATTR`), validates there are
    no name collisions — neither among the decorated methods themselves nor
    against tools already registered in `agent.tool_manager` — and reifies
    each into an `AgentMethodTool`.

    The returned exposure set is a plain list. It is **never** registered
    into `agent.tool_manager` (OQ2) — callers (the MCP mount, TASK-2602)
    are responsible for what they do with it.

    Args:
        agent: The agent instance to scan. Must expose `tool_manager` if it
            has any `@mcp_tool`-decorated method (used for the collision
            check); agents with none are never asked for it.

    Returns:
        The agent's exposure set — one `AgentMethodTool` per decorated
        method. Empty if the agent declares none.

    Raises:
        ValueError: If two decorated methods declare the same MCP tool
            name, or a decorated name collides with an existing
            `agent.tool_manager` tool. The message names the agent class
            and the offending method(s).
    """
    agent_cls_name = type(agent).__name__
    declared_by_method: dict[str, MCPToolDeclaration] = {}
    name_to_method: dict[str, str] = {}

    for method_name, member in inspect.getmembers(type(agent), predicate=inspect.iscoroutinefunction):
        declaration = getattr(member, MCP_TOOL_ATTR, None)
        if declaration is None:
            continue
        if declaration.name in name_to_method:
            raise ValueError(
                f"agent {agent_cls_name!r}: duplicate @mcp_tool name "
                f"{declaration.name!r} declared on methods "
                f"{name_to_method[declaration.name]!r} and {method_name!r}"
            )
        name_to_method[declaration.name] = method_name
        declared_by_method[method_name] = declaration

    if not declared_by_method:
        return []

    existing_tool_names = set(agent.tool_manager.list_tools())
    exposure_set: list[AgentMethodTool] = []
    for method_name, declaration in declared_by_method.items():
        if declaration.name in existing_tool_names:
            raise ValueError(
                f"agent {agent_cls_name!r}: @mcp_tool name {declaration.name!r} "
                f"on method {method_name!r} collides with an existing "
                "tool_manager tool"
            )
        exposure_set.append(AgentMethodTool(agent, method_name, declaration))

    return exposure_set

mcp_tool

mcp_tool(*, name: str, description: str, args_schema: type[BaseModel], returns: type[BaseModel], scope: str, read_only_hint: bool = False, idempotent_hint: bool = False, requires_confirmation: bool = False, max_result_tokens: int | None = None) -> Callable[[F], F]

Mark a bound agent method as externally callable over MCP.

Marks only. Reification into an AgentMethodTool happens at configure() time (TASK-2600). The decorated method is NEVER registered into the owning agent's ToolManager and does not become LLM-callable inside that agent (spec OQ2 — the single most important invariant of this feature).

PARAMETER DESCRIPTION
name

MCP tool name exposed to clients.

TYPE: str

description

Human-readable tool description.

TYPE: str

args_schema

Pydantic model describing the call arguments. Mandatory — no schema inference in v1.

TYPE: type[BaseModel]

returns

Pydantic model describing the return payload. Mandatory.

TYPE: type[BaseModel]

scope

PBAC action/resource scope enforced at call time.

TYPE: str

read_only_hint

MCP readOnlyHint annotation. Defaults to False.

TYPE: bool DEFAULT: False

idempotent_hint

MCP idempotentHint annotation. Defaults to False.

TYPE: bool DEFAULT: False

requires_confirmation

Whether MCP callers must pass confirm=true; maps to routing_meta["requires_confirmation"] / destructiveHint. Defaults to False.

TYPE: bool DEFAULT: False

max_result_tokens

Per-tool result-size cap overriding the mount default. Defaults to None.

TYPE: int | None DEFAULT: None

RETURNS DESCRIPTION
Callable[[F], F]

A decorator that attaches an MCPToolDeclaration to the decorated

Callable[[F], F]

async method and returns it unchanged.

RAISES DESCRIPTION
TypeError

If the decorated callable is not an async def method, if a mandatory field is missing/empty, or if args_schema / returns is not a BaseModel subclass.

Source code in packages/ai-parrot/src/parrot/mcp/agent_tools.py
def mcp_tool(
    *,
    name: str,
    description: str,
    args_schema: type[BaseModel],
    returns: type[BaseModel],
    scope: str,
    read_only_hint: bool = False,
    idempotent_hint: bool = False,
    requires_confirmation: bool = False,
    max_result_tokens: int | None = None,
) -> Callable[[F], F]:
    """Mark a bound agent method as externally callable over MCP.

    Marks only. Reification into an ``AgentMethodTool`` happens at
    ``configure()`` time (TASK-2600). The decorated method is NEVER
    registered into the owning agent's ``ToolManager`` and does not become
    LLM-callable inside that agent (spec OQ2 — the single most important
    invariant of this feature).

    Args:
        name: MCP tool name exposed to clients.
        description: Human-readable tool description.
        args_schema: Pydantic model describing the call arguments. Mandatory
            — no schema inference in v1.
        returns: Pydantic model describing the return payload. Mandatory.
        scope: PBAC action/resource scope enforced at call time.
        read_only_hint: MCP ``readOnlyHint`` annotation. Defaults to `False`.
        idempotent_hint: MCP ``idempotentHint`` annotation. Defaults to `False`.
        requires_confirmation: Whether MCP callers must pass `confirm=true`;
            maps to `routing_meta["requires_confirmation"]` /
            `destructiveHint`. Defaults to `False`.
        max_result_tokens: Per-tool result-size cap overriding the mount
            default. Defaults to `None`.

    Returns:
        A decorator that attaches an `MCPToolDeclaration` to the decorated
        async method and returns it unchanged.

    Raises:
        TypeError: If the decorated callable is not an `async def` method,
            if a mandatory field is missing/empty, or if `args_schema` /
            `returns` is not a `BaseModel` subclass.
    """

    def decorator(fn: F) -> F:
        if not inspect.iscoroutinefunction(fn):
            raise TypeError(f"@mcp_tool requires an async method: {fn.__qualname__}")
        declaration = MCPToolDeclaration(
            name=name,
            description=description,
            args_schema=args_schema,
            returns=returns,
            scope=scope,
            read_only_hint=read_only_hint,
            idempotent_hint=idempotent_hint,
            requires_confirmation=requires_confirmation,
            max_result_tokens=max_result_tokens,
        )
        setattr(fn, MCP_TOOL_ATTR, declaration)
        return fn

    return decorator