MCP Reference

sdk mcp model-context-protocol integrations external-tools

MCP Reference

The ctx.mcp API provides access to external MCP (Model Context Protocol) servers configured in your pack manifest. MCP servers expose tools that become callable from your pack commands—no LLM required.

Overview

MCP integration allows your pack to:

  • Call external tools from MCP servers (GitHub, PostgreSQL, Slack, etc.)
  • Discover available tools and their schemas at runtime
  • Use a unified interface regardless of the underlying MCP server

Note: In code examples, Args represents your Pydantic model for command arguments (e.g., CreateIssueArgs). See Commands Reference for defining argument models.

from huitzo_sdk import command, Context

@command("create-issue", namespace="mypack")
async def create_issue(args: Args, ctx: Context) -> dict:
    """Create a GitHub issue using MCP."""
    result = await ctx.mcp.call(
        server="github",
        tool="create_issue",
        arguments={
            "owner": args.owner,
            "repo": args.repo,
            "title": args.title,
            "body": args.body
        }
    )
    return {"issue_number": result["number"], "url": result["html_url"]}

Quick Start

1. Configure MCP Server in Manifest

# huitzo.yaml
mcp_servers:
  - name: github
    type: stdio
    command: ["uvx", "mcp-server-github"]
    env:
      GITHUB_TOKEN: "${secrets.GITHUB_TOKEN}"

secrets:
  user_required:
    - name: "GITHUB_TOKEN"
      description: "Your GitHub personal access token"
      help_url: "https://github.com/settings/tokens"

2. Call MCP Tools in Commands

from huitzo_sdk import command, Context

@command("list-repos", namespace="devtools")
async def list_repos(args: Args, ctx: Context) -> dict:
    """List GitHub repositories."""
    result = await ctx.mcp.call(
        server="github",
        tool="list_repositories",
        arguments={"owner": args.owner}
    )
    return {"repositories": result["repos"]}

3. Run Your Pack

# Start development session
huitzo pack dev

# Test the command
curl -X POST http://localhost:8080/api/v1/commands/devtools/list-repos \
  -H "Content-Type: application/json" \
  -d '{"owner": "acme"}'

ctx.mcp API Reference

ctx.mcp.call()

Execute a tool on an MCP server.

Signature:

async def call(
    server: str,
    tool: str,
    arguments: dict,
    timeout: int | None = None
) -> dict

Parameters:

Parameter Type Required Default Description
server str Yes — Name of the MCP server (from manifest)
tool str Yes — Name of the tool to execute
arguments dict Yes — Tool arguments (validated against schema)
timeout int No 30 Timeout in seconds

Returns: dict — Tool result (structure depends on the tool)

Raises: - MCPConnectionError — Server unreachable - MCPToolError — Tool execution failed - MCPTimeoutError — Execution exceeded timeout - MCPSchemaError — Arguments don't match schema

Example:

@command("query-db", namespace="data")
async def query_db(args: Args, ctx: Context) -> dict:
    result = await ctx.mcp.call(
        server="postgres",
        tool="query",
        arguments={
            "sql": "SELECT * FROM users WHERE active = true",
            "params": []
        },
        timeout=60  # Allow 60 seconds for large queries
    )
    return {"rows": result["rows"], "count": len(result["rows"])}

ctx.mcp.list_tools()

List available tools from one or all MCP servers.

Signature:

async def list_tools(server: str | None = None) -> list[MCPTool]

Parameters:

Parameter Type Required Default Description
server str No None Server name, or None for all servers

Returns: list[MCPTool] — List of available tools

Example:

@command("discover-tools", namespace="admin")
async def discover_tools(args: Args, ctx: Context) -> dict:
    # List tools from specific server
    github_tools = await ctx.mcp.list_tools("github")

    # List tools from all configured servers
    all_tools = await ctx.mcp.list_tools()

    return {
        "github_tools": [t.name for t in github_tools],
        "all_tools": [{"server": t.server, "name": t.name} for t in all_tools]
    }

ctx.mcp.get_tool_schema()

Get the JSON Schema for a specific tool.

Signature:

async def get_tool_schema(server: str, tool: str) -> dict

Parameters:

Parameter Type Required Default Description
server str Yes — Server name
tool str Yes — Tool name

Returns: dict — JSON Schema for the tool's input

Example:

@command("inspect-tool", namespace="admin")
async def inspect_tool(args: Args, ctx: Context) -> dict:
    schema = await ctx.mcp.get_tool_schema("github", "create_issue")
    return {
        "tool": "create_issue",
        "required_fields": schema.get("required", []),
        "properties": list(schema.get("properties", {}).keys())
    }

ctx.mcp.servers

Property that returns information about configured MCP servers.

servers = ctx.mcp.servers  # dict[str, MCPServerInfo]

Returns: dict[str, MCPServerInfo] — Server configurations

Example:

@command("list-servers", namespace="admin")
async def list_servers(args: Args, ctx: Context) -> dict:
    return {
        "servers": [
            {
                "name": name,
                "type": info.type,
                "connected": info.connected
            }
            for name, info in ctx.mcp.servers.items()
        ]
    }

Discovering Tools

CLI Commands (Planned)

Note: These CLI commands will be implemented alongside the MCP runtime. They are documented here for reference and will be added to the CLI Reference when available.

# List configured MCP servers
huitzo mcp list-servers

# List tools from a specific server
huitzo mcp list-tools github

# Get tool schema
huitzo mcp list-tools github --tool create_issue --schema

# Test a tool directly
huitzo mcp call github create_issue '{"owner": "acme", "repo": "test", "title": "Test"}'

Programmatic Discovery

@command("explore-mcp", namespace="admin")
async def explore_mcp(args: Args, ctx: Context) -> dict:
    """Explore available MCP tools."""
    servers = ctx.mcp.servers
    result = {"servers": {}}

    for server_name in servers:
        tools = await ctx.mcp.list_tools(server_name)
        result["servers"][server_name] = {
            "tool_count": len(tools),
            "tools": [
                {
                    "name": t.name,
                    "description": t.description,
                    "required_args": t.required_args
                }
                for t in tools
            ]
        }

    return result

Error Handling

MCPError Hierarchy

MCPError (base)
├── MCPConnectionError    # Server unreachable
├── MCPToolError          # Tool execution failed
├── MCPTimeoutError       # Timeout exceeded
└── MCPSchemaError        # Invalid arguments

Handling Errors

from huitzo_sdk import command, Context
from huitzo_sdk.errors import (
    MCPError,
    MCPConnectionError,
    MCPToolError,
    MCPTimeoutError,
    MCPSchemaError
)

@command("robust-github", namespace="devtools")
async def robust_github(args: Args, ctx: Context) -> dict:
    """Create issue with comprehensive error handling."""
    try:
        result = await ctx.mcp.call(
            server="github",
            tool="create_issue",
            arguments={
                "owner": args.owner,
                "repo": args.repo,
                "title": args.title
            }
        )
        return {"success": True, "issue": result}

    except MCPConnectionError as e:
        ctx.log.error(f"GitHub MCP server unavailable: {e.message}")
        return {
            "success": False,
            "error": "GitHub integration temporarily unavailable",
            "retry": True
        }

    except MCPSchemaError as e:
        ctx.log.error(f"Invalid arguments: {e.message}")
        return {
            "success": False,
            "error": f"Invalid input: {e.field}",
            "validation_errors": e.errors
        }

    except MCPToolError as e:
        ctx.log.error(f"GitHub tool failed: {e.message}")
        return {
            "success": False,
            "error": f"GitHub error: {e.message}",
            "tool_error_code": e.code
        }

    except MCPTimeoutError as e:
        ctx.log.error(f"GitHub call timed out after {e.timeout}s")
        return {
            "success": False,
            "error": "Request timed out",
            "timeout": e.timeout
        }

Error Attributes

MCPConnectionError: - server — Server name that failed to connect - message — Error details

MCPToolError: - server — Server name - tool — Tool name that failed - code — Error code from the MCP server - message — Error message

MCPTimeoutError: - server — Server name - tool — Tool name - timeout — Configured timeout value

MCPSchemaError: - server — Server name - tool — Tool name - field — Field that failed validation - errors — List of validation errors - message — Human-readable message

Type Reference

MCPTool

Information about an MCP tool.

class MCPTool:
    name: str                    # Tool name
    description: str             # Human-readable description
    server: str                  # Server that provides this tool
    input_schema: dict           # JSON Schema for arguments
    required_args: list[str]     # Required argument names

MCPServerInfo

Information about a configured MCP server.

class MCPServerInfo:
    name: str                    # Server name
    type: str                    # "stdio" or "http"
    connected: bool              # Whether currently connected
    tool_count: int              # Number of available tools

Best Practices

1. Validate Tool Existence

# ✅ Good: Check before calling
tools = await ctx.mcp.list_tools("github")
tool_names = [t.name for t in tools]

if "create_issue" not in tool_names:
    raise CommandError(
        message="GitHub MCP server doesn't support create_issue",
        details={"available_tools": tool_names}
    )

2. Use Appropriate Timeouts

# ✅ Good: Short timeout for fast operations
user = await ctx.mcp.call("github", "get_user", {"username": "octocat"}, timeout=10)

# ✅ Good: Long timeout for slow operations
result = await ctx.mcp.call("postgres", "query", {"sql": complex_query}, timeout=120)

3. Handle Partial Failures

@command("multi-server", namespace="workflow")
async def multi_server(args: Args, ctx: Context) -> dict:
    results = {"github": None, "slack": None, "errors": []}

    # Try GitHub
    try:
        results["github"] = await ctx.mcp.call("github", "create_issue", {...})
    except MCPError as e:
        results["errors"].append({"server": "github", "error": str(e)})

    # Try Slack even if GitHub failed
    try:
        results["slack"] = await ctx.mcp.call("slack", "post_message", {...})
    except MCPError as e:
        results["errors"].append({"server": "slack", "error": str(e)})

    return results

4. Log Tool Usage for Debugging

# ✅ Good: Log before and after
ctx.log.info("Calling GitHub create_issue", extra={
    "owner": args.owner,
    "repo": args.repo
})

result = await ctx.mcp.call("github", "create_issue", {...})

ctx.log.info("GitHub issue created", extra={
    "issue_number": result.get("number"),
    "url": result.get("html_url")
})

5. Wrap Tools for Business Logic

async def create_labeled_issue(ctx: Context, owner: str, repo: str, title: str, labels: list[str]) -> dict:
    """Business logic wrapper around MCP tool."""
    # Create issue
    issue = await ctx.mcp.call("github", "create_issue", {
        "owner": owner,
        "repo": repo,
        "title": title
    })

    # Add labels (if the tool supports it)
    if labels:
        await ctx.mcp.call("github", "add_labels", {
            "owner": owner,
            "repo": repo,
            "issue_number": issue["number"],
            "labels": labels
        })

    return issue

@command("create-bug", namespace="devtools")
async def create_bug(args: Args, ctx: Context) -> dict:
    return await create_labeled_issue(
        ctx, args.owner, args.repo, args.title,
        labels=["bug", "needs-triage"]
    )