Building MCP Servers with Chinese AI: Model Context Protocol Guide

Environment Setup

# Install MCP SDK
pip install mcp

# Set your AIWave API key
export AIWAVE_API_KEY="your-api-key-here"

Server 1: Filesystem Tools

Let's build an MCP server that lets the AI model read, search, and analyze files:

"""MCP Server: Filesystem Tools for AI Models"""
import os
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server

server = Server("filesystem-tools")

@server.list_tools()
async def list_tools():
    return [
        {
            "name": "read_file",
            "description": "Read contents of a file",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "File path"},
                    "max_lines": {"type": "integer", "description": "Max lines to read", "default": 100},
                },
                "required": ["path"],
            },
        },
        {
            "name": "search_files",
            "description": "Search for files matching a pattern",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "directory": {"type": "string", "description": "Directory to search"},
                    "pattern": {"type": "string", "description": "Glob pattern (e.g. '*.py')"},
                },
                "required": ["directory", "pattern"],
            },
        },
        {
            "name": "grep_content",
            "description": "Search file contents for a regex pattern",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "directory": {"type": "string"},
                    "pattern": {"type": "string", "description": "Regex pattern"},
                    "file_pattern": {"type": "string", "description": "File filter", "default": "*"},
                },
                "required": ["directory", "pattern"],
            },
        },
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "read_file":
        path = arguments["path"]
        max_lines = arguments.get("max_lines", 100)
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            lines = f.readlines()[:max_lines]
        return f"Contents of {path} ({len(lines)} lines):\n" + "".join(lines)
    
    elif name == "search_files":
        import glob
        matches = glob.glob(os.path.join(arguments["directory"], arguments["pattern"]))
        return f"Found {len(matches)} files:\n" + "\n".join(matches[:50])
    
    elif name == "grep_content":
        import re
        results = []
        pattern = re.compile(arguments["pattern"])
        for root, dirs, files in os.walk(arguments["directory"]):
            for f in files:
                if not f.endswith(('.py', '.js', '.ts', '.md', '.txt')):
                    continue
                try:
                    path = os.path.join(root, f)
                    with open(path, 'r', errors='replace') as fh:
                        for i, line in enumerate(fh, 1):
                            if pattern.search(line):
                                results.append(f"{path}:{i}: {line.strip()}")
                                if len(results) > 50:
                                    return "\n".join(results) + "\n... (truncated)"
                except: pass
        return "\n".join(results) if results else "No matches found"

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())

Server 2: Database Query

"""MCP Server: Database Query Tools"""
import sqlite3
from mcp.server import Server
from mcp.server.stdio import stdio_server

server = Server("db-tools")
DB_PATH = "app.db"

@server.list_tools()
async def list_tools():
    return [
        {
            "name": "list_tables",
            "description": "List all tables in the database",
            "inputSchema": {"type": "object", "properties": {}},
        },
        {
            "name": "query",
            "description": "Execute a read-only SQL query. Only SELECT statements allowed.",
            "inputSchema": {
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "SELECT SQL query"},
                },
                "required": ["sql"],
            },
        },
        {
            "name": "describe_table",
            "description": "Show table schema (columns and types)",
            "inputSchema": {
                "type": "object",
                "properties": {"table": {"type": "string"}},
                "required": ["table"],
            },
        },
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    
    if name == "list_tables":
        tables = conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
        return "Tables: " + ", ".join(t["name"] for t in tables)
    
    elif name == "describe_table":
        columns = conn.execute(f"PRAGMA table_info({arguments['table']})").fetchall()
        return "\n".join(f"  {c['name']} ({c['type']})" for c in columns)
    
    elif name == "query":
        sql = arguments["sql"].strip()
        if not sql.upper().startswith("SELECT"):
            return "Error: Only SELECT queries are allowed for safety."
        try:
            rows = conn.execute(sql).fetchall()
            if not rows:
                return "Query returned 0 rows."
            headers = list(rows[0].keys())
            result = " | ".join(headers) + "\n" + "-" * 60 + "\n"
            for row in rows[:100]:
                result += " | ".join(str(v) for v in row) + "\n"
            if len(rows) > 100:
                result += f"... ({len(rows) - 100} more rows)"
            return result
        except Exception as e:
            return f"SQL Error: {e}"

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())

Server 3: REST API Bridge

"""MCP Server: REST API Bridge - Let AI models call any HTTP API"""
import json
import urllib.request
from mcp.server import Server
from mcp.server.stdio import stdio_server

server = Server("api-bridge")

@server.list_tools()
async def list_tools():
    return [{
        "name": "http_request",
        "description": "Make an HTTP GET request and return the response body",
        "inputSchema": {
            "type": "object",
            "properties": {
                "url": {"type": "string", "description": "URL to fetch"},
                "headers": {"type": "object", "description": "Request headers (optional)"},
                "max_bytes": {"type": "integer", "description": "Max response bytes", "default": 10000},
            },
            "required": ["url"],
        },
    }]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    if name != "http_request":
        return "Unknown tool"
    
    url = arguments["url"]
    headers = arguments.get("headers", {})
    max_bytes = arguments.get("max_bytes", 10000)
    
    req = urllib.request.Request(url, headers=headers)
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            body = resp.read(max_bytes).decode("utf-8", errors="replace")
            # Truncate JSON responses for token efficiency
            try:
                data = json.loads(body)
                body = json.dumps(data, indent=2)[:max_bytes]
            except: pass
            return f"Status: {resp.status}\nContent ({len(body)} chars):\n{body}"
    except Exception as e:
        return f"HTTP Error: {e}"

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())

Connecting to DeepSeek V4 Pro

Now connect these MCP servers to DeepSeek V4 Pro using the AIWave API:

import os, json, subprocess
import openai

client = openai.OpenAI(
    api_key=os.environ["AIWAVE_API_KEY"],
    base_url="https://aiwave.live/v1",
)

def run_mcp_server(server_script: str, input_text: str) -> str:
    """Launch an MCP server, send input, capture output."""
    proc = subprocess.run(
        ["python", server_script],
        input=input_text, capture_output=True, text=True, timeout=30
    )
    return proc.stdout or proc.stderr

def agent_with_tools(query: str, tools: list, max_turns: int = 8) -> str:
    """Agent loop with MCP-style tool calling via AIWave."""
    messages = [{"role": "user", "content": query}]
    
    for _ in range(max_turns):
        resp = client.chat.completions.create(
            model="deepseek-v4-pro",
            messages=messages,
            tools=tools,
            temperature=0.1,
        )
        msg = resp.choices[0].message
        messages.append(msg)
        
        if not msg.tool_calls:
            return msg.content
        
        for tc in msg.tool_calls:
            args = json.loads(tc.function.arguments)
            # Execute tool locally
            result = local_tool_dispatch(tc.function.name, args)
            messages.append({
                "role": "tool",
                "tool_call_id": tc.id,
                "content": json.dumps(result),
            })
    return "Max turns reached."

Cost Analysis