Sales Tools MCP Server

Building an MCP Server from Scratch: Exposing Your Python Tools to Any AI Agent

Every AI agent framework has its own way of defining tools. You write a function, decorate it, and your agent can call it. But what happens when you want a different agent, running in a different framework, to use the same tool? You rewrite it. The Model Context Protocol (MCP) solves this by defining a universal standard: build your tools once as an MCP server, and any MCP-compatible client can discover and use them automatically.

In this project we build a custom MCP server that exposes business tools (SQL queries, KPI calculations, report generation) over a sales database. Then we consume it from a Strands Agent using the native MCPClient integration. The same server could be used from Claude Code, Cursor, Amazon Q, or any other MCP-compatible client without changing a single line.

graph LR
    CLI["CLI (Click)"] --> Agent["Strands Agent (Claude Sonnet)"]
    Agent --> MCPClient["MCPClient (stdio)"]
    MCPClient --> Server["MCP Server (FastMCP)"]
    Server --> SQL["sql_query"]
    Server --> KPI["get_kpi"]
    Server --> Report["sales_report"]
    SQL --> DB["SQLite"]
    KPI --> DB
    Report --> DB

What is MCP?

The Model Context Protocol is an open standard that defines how AI agents communicate with external tools. Instead of each framework inventing its own tool interface, MCP provides a shared protocol: the server exposes tools with JSON schemas describing their parameters, and the client discovers and invokes them over a transport layer (stdio, HTTP, or SSE).

The key insight is separation of concerns. The MCP server knows nothing about the AI model consuming it. The client knows nothing about the tool implementation. They communicate through a well-defined contract, just like a REST API, but designed for agent-tool interaction.

Building the MCP server

The server uses FastMCP, the high-level Python API included in the official MCP SDK. Each tool is a decorated function where type hints and docstrings automatically generate the JSON schema that clients use to understand parameters:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(name="sales-tools")

@mcp.tool()
def sql_query(query: str) -> str:
    """Execute a SQL query against the sales database.

    The database contains three tables:
    - customers: id, name, region, created_at
    - products: id, name, category, unit_price
    - orders: id, customer_id, product_id, quantity, total_amount, order_date

    Only SELECT queries are allowed.
    """
    return query_sales(query)

@mcp.tool()
def get_kpi(metric: str, period: str = "all") -> str:
    """Calculate a business KPI from the sales database.

    Available metrics: revenue, order_count, avg_order_value, top_products,
    top_customers, revenue_by_region, revenue_by_category.

    Available periods: all, last_month, last_quarter, last_year, 2024, 2025, 2026.
    """
    return calculate_kpi(metric, period)

@mcp.tool()
def sales_report() -> str:
    """Generate a comprehensive markdown sales report with key metrics,
    top products, regional breakdown, and category analysis."""
    return generate_report()

if __name__ == "__main__":
    mcp.run()

That's the entire server. The @mcp.tool() decorator does three things: registers the function as an MCP tool, extracts parameter types from the signature to build the JSON schema, and uses the docstring as the tool description that agents read to decide when and how to call it. When you run python server/main.py, it starts listening on stdio for MCP requests.

The business tools

Behind the MCP layer, each tool is a regular Python function operating on a SQLite database.

The SQL query tool validates that only SELECT statements are allowed (no accidental data mutations) and returns results as JSON. If the query fails, it returns the error along with the schema description so the agent can self-correct:

def query_sales(sql: str) -> str:
    normalized = sql.strip().upper()
    if not normalized.startswith("SELECT"):
        return json.dumps({"error": "Only SELECT queries are allowed"})

    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    try:
        cursor = conn.execute(sql)
        columns = [desc[0] for desc in cursor.description] if cursor.description else []
        rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
        return json.dumps(rows, default=str)
    except sqlite3.Error as e:
        return json.dumps({"error": str(e), "schema": SCHEMA_DESCRIPTION})
    finally:
        conn.close()

The KPI tool uses a Strategy pattern to route metric names to specific calculators. Each calculator runs an optimized SQL query and returns structured results:

def _calculate(conn: sqlite3.Connection, metric: str, date_filter: str) -> dict:
    calcula