M07: MCP & Rich Output

# ucc_server.py — an MCP server.  Install:  pip install "mcp[cli]" httpx
# WHAT: Exposes tools (search_filings, web_search) and a resource (filing files)
# WHY:  FastMCP turns typed functions into protocol-correct MCP primitives
# GOTCHA: Run over stdio for local dev; the host launches this file as a subprocess
import json, pathlib, httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("ucc-tools")                       # the server's name, shown to clients
FILINGS = pathlib.Path("filings")                # a folder of mock UCC filing .txt files

@mcp.tool()
def search_filings(query: str, limit: int = 5) -> list[dict]:
    """Search local UCC filings for a keyword. Returns matching filings with a snippet.

    Args:
        query: keyword to search for (case-insensitive)
        limit: max number of results to return
    """
    results = []
    for path in sorted(FILINGS.glob("*.txt")):
        text = path.read_text(encoding="utf-8")
        if query.lower() in text.lower():
            idx = text.lower().index(query.lower())
            results.append({"file": path.name, "snippet": text[max(0, idx-40):idx+60]})
        if len(results) >= limit:
            break
    return results

@mcp.tool()
def web_search(query: str, k: int = 3) -> list[dict]:
    """Search the public web for up-to-date information the model can't know.

    Args:
        query: the search query
        k: number of results to return
    """
    # AI-native web search returns clean, agent-ready snippets — not a page of HTML.
    # Swap this for any provider (Brave, Tavily, SearXNG self-hosted, etc.).
    try:
        resp = httpx.get("https://api.search.example/q",
                         params={"q": query, "k": k}, timeout=10)
        resp.raise_for_status()
        return resp.json()["results"]            # [{title, url, snippet}, ...]
    except httpx.HTTPError as e:
        return [{"error": f"web_search unavailable: {e}"}]

@mcp.resource("filing://{name}")
def read_filing(name: str) -> str:
    """Return the full text of one filing by file name (a READ-ONLY resource)."""
    path = FILINGS / name
    if not path.is_file():
        raise FileNotFoundError(f"no filing named {name!r}")
    return path.read_text(encoding="utf-8")

if __name__ == "__main__":
    mcp.run()                                    # default transport: stdio

# bridge.py — let a local Mistral agent use MCP tools.  pip install "mcp[cli]" openai
# WHAT: Discover MCP tools, expose them to Mistral, dispatch tool_calls back to MCP
# WHY:  Open models speak OpenAI tool_calls, not MCP — this is the translation layer
# GOTCHA: Tool results go back as role:"tool" messages keyed by tool_call_id
import asyncio, json
from openai import OpenAI
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

llm = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")

def mcp_tool_to_openai(t) -> dict:
    # Map an MCP tool's schema onto the OpenAI tool-calling format Mistral expects
    return {"type": "function", "function": {
        "name": t.name,
        "description": t.description or "",
        "parameters": t.inputSchema,            # MCP already gives JSON Schema
    }}

async def run(user_msg: str) -> str:
    # Launch the MCP server as a subprocess and open a session over stdio
    params = StdioServerParameters(command="python", args=["ucc_server.py"])
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()                       # MCP handshake
            tools = (await session.list_tools()).tools       # tools/list
            openai_tools = [mcp_tool_to_openai(t) for t in tools]

            messages = [{"role": "user", "content": user_msg}]
            for _ in range(6):                               # bounded agent loop
                resp = llm.chat.completions.create(
                    model="mistral", messages=messages, tools=openai_tools)
                msg = resp.choices[0].message
                if not msg.tool_calls:                       # model is done
                    return msg.content
                messages.append(msg)
                for call in msg.tool_calls:                  # dispatch each call to MCP
                    args = json.loads(call.function.arguments or "{}")
                    result = await session.call_tool(call.function.name, args)  # tools/call
                    text = "".join(c.text for c in result.content if c.type == "text")
                    messages.append({"role": "tool", "tool_call_id": call.id,
                                     "content": text})        # reply keyed by id
            return "stopped: hit the tool-call iteration cap"

if __name__ == "__main__":
    print(asyncio.run(run("Find filings that mention bankruptcy and summarize them.")))

# add to ucc_server.py — a tool that returns a CHART, not text.  pip install matplotlib
# WHAT: Render a bar chart from labelled values and return it as an MCP image
# WHY:  Lets the agent answer "show the trend" with an actual visual artifact
# GOTCHA: Use the non-interactive 'Agg' backend — there is no display on a server
import io, base64
import matplotlib
matplotlib.use("Agg")                            # headless: render to a buffer, no GUI
import matplotlib.pyplot as plt
from mcp.server.fastmcp import Image

@mcp.tool()
def make_bar_chart(labels: list[str], values: list[float], title: str = "") -> Image:
    """Render a bar chart and return it as a PNG image.

    Args:
        labels: x-axis category labels (e.g. months)
        values: the bar heights, one per label
        title: chart title
    """
    if len(labels) != len(values):
        raise ValueError("labels and values must be the same length")
    fig, ax = plt.subplots(figsize=(6, 3.5))
    ax.bar(labels, values, color="#F97316")
    ax.set_title(title)
    fig.tight_layout()

    buf = io.BytesIO()
    fig.savefig(buf, format="png", dpi=120)      # render to memory, not disk
    plt.close(fig)                               # always free the figure
    # MCP Image carries the bytes; the host renders it inline. No file path leaks.
    return Image(data=buf.getvalue(), format="png")

mkdir mcp-lab && cd mcp-lab
pip install "mcp[cli]" openai matplotlib httpx
ollama pull mistral
mkdir filings
printf 'Debtor: Acme LLC\nSecured party: First Bank\nNote: bankruptcy risk flagged Q2.\n' > filings/f-101.txt
printf 'Debtor: Globex Inc\nSecured party: Credit Union\nNote: clean standing.\n'           > filings/f-102.txt
printf 'Debtor: Initech\nSecured party: First Bank\nNote: prior bankruptcy on record.\n'    > filings/f-103.txt