Biến CLI thành MCP Server: agent tự discover và invoke tool của bạn

# mcp_server.py
import json
import os
from dataclasses import asdict
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
from adapters.user_adapter import UserAdapter, AdapterError
 
server = Server("user-management-mcp")
 
 
def get_adapter() -> UserAdapter:
    token = os.environ.get("API_TOKEN", "")
    return UserAdapter(token=token)
 
 
@server.list_tools()
async def list_tools() -> list[types.Tool]:
    """Declare available tools — agent sẽ receive list này khi connect."""
    return [
        types.Tool(
            name="list_users",
            description="List all users. Supports filtering by role.",
            inputSchema={
                "type": "object",
                "properties": {
                    "role": {
                        "type": "string",
                        "enum": ["admin", "member", "viewer"],
                        "description": "Filter by role (optional)",
                    }
                },
            },
        ),
        types.Tool(
            name="get_user",
            description="Get a specific user by ID.",
            inputSchema={
                "type": "object",
                "properties": {
                    "user_id": {"type": "integer", "description": "User ID"}
                },
                "required": ["user_id"],
            },
        ),
        types.Tool(
            name="ensure_user",
            description="Create user if not exists. Idempotent — safe to call multiple times.",
            inputSchema={
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "email": {"type": "string", "format": "email"},
                    "role": {"type": "string", "enum": ["admin", "member", "viewer"], "default": "member"},
                },
                "required": ["name", "email"],
            },
        ),
    ]
 
 
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    """Execute tool và trả kết quả dưới dạng JSON string."""
    try:
        with get_adapter() as adapter:
            if name == "list_users":
                users = list(adapter.list_users(role=arguments.get("role")))
                result = [asdict(u) for u in users]
            elif name == "get_user":
                user = adapter.get_user(arguments["user_id"])
                result = asdict(user)
            elif name == "ensure_user":
                user, action = adapter.ensure_user(
                    name=arguments["name"],
                    email=arguments["email"],
                    role=arguments.get("role", "member"),
                )
                result = {"action": action, "user": asdict(user)}
            else:
                result = {"error": "unknown_tool", "tool": name}
 
    except AdapterError as e:
        result = {"error": e.code, "message": str(e), "exit_code": e.exit_code}
 
    return [types.TextContent(type="text", text=json.dumps(result, indent=2))]
 
 
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())
 
 
if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

{
  "mcpServers": {
    "user-management": {
      "command": "python",
      "args": ["mcp_server.py"],
      "env": {
        "API_TOKEN": "${API_TOKEN}"
      }
    }
  }
}

{# templates/mcp_server.py.j2 #}
"""Auto-generated MCP server. DO NOT EDIT MANUALLY."""
import json
import os
from dataclasses import asdict
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
{% for resource in resources %}
from adapters.{{ resource }}_adapter import {{ resource | title }}Adapter, AdapterError
{% endfor %}
 
server = Server("{{ api_name }}-mcp")
 
TOOLS = [
    {% for method in methods %}
    types.Tool(
        name="{{ method.id | replace('.', '_') }}",
        description="{{ method.summary }}",
        inputSchema={
            "type": "object",
            "properties": {
                {% for param in method.parameters %}
                "{{ param.name }}": {
                    "type": "{{ param.type }}",
                    {% if param.enum %}"enum": {{ param.enum | tojson }},{% endif %}
                    "description": "{{ param.description }}",
                },
                {% endfor %}
            },
            "required": [{% for p in method.parameters if p.required %}"{{ p.name }}"{% if not loop.last %}, {% endif %}{% endfor %}],
        },
    ),
    {% endfor %}
]

codegen: extract normalize
	python pipeline/generator.py $(METADATA) $(GENERATED) $(TEMPLATES)
	python pipeline/generate_mcp.py $(METADATA) mcp_server.py  # ← new

# agent_workflow.py
import subprocess
import json
from typing import Any
 
 
def call_cli(command: list[str]) -> dict:
    """Gọi CLI tool — zero token overhead, sync."""
    result = subprocess.run(command + ["--json"], capture_output=True, text=True)
    if result.returncode != 0:
        error = json.loads(result.stderr) if result.stderr else {"error": "unknown"}
        return {"success": False, **error, "exit_code": result.returncode}
    return {"success": True, "data": json.loads(result.stdout)}
 
 
# Simple lookup → CLI (fast, no MCP overhead)
user = call_cli(["myapp", "get-user", str(user_id)])
 
# Complex workflow → MCP tools (rich discovery, stateful)
# Agent với MCP tools sẽ orchestrate qua nhiều tool calls