Chapter 28: Building Custom MCP Servers#

from tina4_python.mcp import McpServer​mcp = McpServer("/api/my-tools", name="My App Tools", version="1.0.0")

from tina4_python.core.router import get, postmcp.register_routes(__import__("tina4_python.core.router", fromlist=["router"]))

from tina4_python.mcp import McpServer, mcp_tool​mcp = McpServer("/crm/mcp", name="CRM Tools")​@mcp_tool("lookup_customer", description="Find a customer by email", server=mcp)def lookup_customer(email: str):    """Search the customer database by email address."""    return db.fetch_one("SELECT * FROM customers WHERE email = ?", [email])​@mcp_tool("recent_orders", description="Get recent orders for a customer", server=mcp)def recent_orders(customer_id: int, limit: int = 10):    """Fetch the most recent orders for a customer."""    result = db.fetch(        "SELECT * FROM orders WHERE customer_id = ? ORDER BY created_at DESC",        [customer_id], limit=limit    )    return result.to_array()

{  "name": "lookup_customer",  "description": "Find a customer by email",  "inputSchema": {    "type": "object",    "properties": {      "email": {"type": "string"}    },    "required": ["email"]  }}

from tina4_python.mcp import mcp_resource​@mcp_resource("crm://product-catalog", description="All active products", server=mcp)def product_catalog():    return db.fetch("SELECT id, name, price, category FROM products WHERE active = 1").to_array()​@mcp_resource("crm://tax-rates", description="Current tax rates by region", server=mcp)def tax_rates():    return db.fetch("SELECT region, rate FROM tax_rates").to_array()

from tina4_python.mcp import McpServer, mcp_tool​mcp = McpServer("/accounting/mcp", name="Accounting Tools")​class AccountingService:    def __init__(self, db):        self.db = db​    @mcp_tool("invoice_lookup", description="Find an invoice by number", server=mcp)    def lookup(self, invoice_no: str):        return self.db.fetch_one(            "SELECT * FROM invoices WHERE invoice_no = ?", [invoice_no]        )​    @mcp_tool("outstanding_balances", description="List all unpaid invoices", server=mcp)    def balances(self, min_amount: float = 0.0):        return self.db.fetch(            "SELECT * FROM invoices WHERE paid = 0 AND total >= ?", [min_amount]        ).to_array()​    @mcp_tool("monthly_summary", description="Revenue summary for a month", server=mcp)    def summary(self, year: int, month: int):        return self.db.fetch_one(            "SELECT SUM(total) as revenue, COUNT(*) as invoice_count "            "FROM invoices WHERE strftime('%Y', created_at) = ? "            "AND strftime('%m', created_at) = ?",            [str(year), f"{month:02d}"]        )​# Create the service instance - methods are already registered via decoratorsaccounting = AccountingService(db)

from tina4_python.core.router import middleware, secured​# Secure the entire MCP path@secured()@middleware(AuthMiddleware)def register_mcp():    mcp.register_routes(router)

@mcp_tool("sensitive_data", description="Access restricted data", server=mcp)def sensitive_data(token: str):    payload = Auth.valid_token_static(token)    if not payload:        return {"error": "Unauthorized"}    return db.fetch("SELECT * FROM sensitive_table").to_array()

# Test the tool function directlydef test_lookup_customer():    result = lookup_customer("alice@example.com")    assert result is not None    assert result["email"] == "alice@example.com"​# Test via MCP protocoldef test_mcp_tool_call():    import json    resp = json.loads(mcp.handle_message({        "jsonrpc": "2.0",        "id": 1,        "method": "tools/call",        "params": {            "name": "lookup_customer",            "arguments": {"email": "alice@example.com"}        }    }))    assert "result" in resp    content = resp["result"]["content"][0]["text"]    assert "alice" in content.lower()

# app.pyfrom tina4_python.core import runfrom tina4_python.orm import bind_databasefrom tina4_python.database import Databasefrom tina4_python.mcp import McpServer, mcp_tool, mcp_resource​db = Database("sqlite:///crm.db")bind_database(db)​# Create MCP servercrm_mcp = McpServer("/crm/mcp", name="CRM Assistant", version="1.0.0")​# Tools@mcp_tool("find_customer", description="Search customers by name or email", server=crm_mcp)def find_customer(query: str):    return db.fetch(        "SELECT * FROM customers WHERE name LIKE ? OR email LIKE ?",        [f"%{query}%", f"%{query}%"]    ).to_array()​@mcp_tool("customer_orders", description="Get all orders for a customer", server=crm_mcp)def customer_orders(customer_id: int):    return db.fetch(        "SELECT o.*, GROUP_CONCAT(oi.product_name) as items "        "FROM orders o LEFT JOIN order_items oi ON o.id = oi.order_id "        "WHERE o.customer_id = ? GROUP BY o.id ORDER BY o.created_at DESC",        [customer_id]    ).to_array()​@mcp_tool("create_note", description="Add a note to a customer record", server=crm_mcp)def create_note(customer_id: int, note: str):    db.insert("customer_notes", {"customer_id": customer_id, "note": note})    return {"success": True}​# Resources@mcp_resource("crm://products", description="Product catalog", server=crm_mcp)def products():    return db.fetch("SELECT * FROM products WHERE active = 1").to_array()​@mcp_resource("crm://stats", description="CRM statistics", server=crm_mcp)def stats():    customers = db.fetch_one("SELECT COUNT(*) as count FROM customers")    orders = db.fetch_one("SELECT COUNT(*) as count, SUM(total) as revenue FROM orders")    return {        "customers": customers["count"],        "orders": orders["count"],        "revenue": orders["revenue"],    }​# Register routesimport tina4_python.core.router as routercrm_mcp.register_routes(router)​if __name__ == "__main__":    run()