Building a Custom MCP Server — The Pathfinder

Why Build Your Own?

Off-the-shelf MCP servers cover popular platforms. When your data or workflow doesn't have a public server, you build one.

The Pathfinder use case: AI agents log their activity into Pathfinder (a Supabase-backed time-tracking app), creating a horizontal timeline showing what each agent did, when, and for how long.

The MCP Python SDK: FastMCP

Python

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("pathfinder")

@mcp.tool()
def my_tool(param: str) -> str:
    """What this tool does. Claude reads this docstring."""
    return f"result: {param}"

mcp.run(transport="stdio")

The @mcp.tool() decorator reads your type hints and docstring to automatically generate the tool definition Claude sees.

The Critical STDIO Rule

Your server communicates via stdin/stdout pipes. NEVER use print() in a stdio MCP server - it corrupts the JSON-RPC protocol.

Python

# ❌ BREAKS YOUR SERVER
print("Processing request")

# ✅ Safe
print("Processing request", file=sys.stderr)
import logging; logging.info("Processing request")

Three Things a Server Can Expose

Tools - functions the LLM calls (main capability):

Python

@mcp.tool()
def log_activity(agent_name: str, task: str) -> str: ...

Resources - read-only data via @ mentions:

Python

@mcp.resource("pathfinder://summary/today")
def todays_summary() -> str: ...

Prompts - slash commands:

Python

@mcp.prompt()
def agent_report(agent_name: str) -> str:
    return f"Summarize today's activity for {agent_name}."

Setup

Bash

# Install uv, then:
uv init pathfinder-mcp
cd pathfinder-mcp
uv add "mcp[cli]" supabase python-dotenv

.env file:

Bash

SUPABASE_URL=https://your-project-id.supabase.co
SUPABASE_KEY=your-service-role-key

Supabase Schema

SQL

create table agent_activity_logs (
  id               uuid default gen_random_uuid() primary key,
  agent_name       text not null,
  task             text not null,
  start_time       timestamptz not null,
  end_time         timestamptz,
  duration_seconds integer,
  status           text check (status in ('running', 'completed', 'failed')),
  notes            text,
  created_at       timestamptz default now()
);
create index on agent_activity_logs (agent_name, start_time);

The Full Server

Python

# pathfinder_mcp.py
import sys, os
from datetime import datetime
from typing import Optional
from mcp.server.fastmcp import FastMCP
from supabase import create_client, Client
from dotenv import load_dotenv

load_dotenv()
mcp = FastMCP("pathfinder")
supabase: Client = create_client(os.environ["SUPABASE_URL"], os.environ["SUPABASE_KEY"])


@mcp.tool()
def log_activity(agent_name: str, task: str, start_time: str,
                 status: str, end_time: Optional[str] = None,
                 notes: Optional[str] = None) -> str:
    """Log an AI agent's activity to Pathfinder.
    Args:
        agent_name: Name of the agent (e.g. 'researcher', 'coder')
        task: Description of what the agent is doing
        start_time: ISO 8601 format (e.g. '2026-02-18T14:30:00Z')
        status: 'running', 'completed', or 'failed'
        end_time: ISO 8601, required if completed/failed
        notes: Optional outcome summary
    """
    data = {"agent_name": agent_name, "task": task, "start_time": start_time, "status": status}
    if end_time:
        data["end_time"] = end_time
        start = datetime.fromisoformat(start_time.replace("Z", "+00:00"))
        end = datetime.fromisoformat(end_time.replace("Z", "+00:00"))
        data["duration_seconds"] = int((end - start).total_seconds())
    if notes:
        data["notes"] = notes
    supabase.table("agent_activity_logs").insert(data).execute()
    return f"Logged: {agent_name} - '{task}' [{status}]"


@mcp.tool()
def update_activity_status(agent_name: str, task: str, end_time: str,
                           status: str, notes: Optional[str] = None) -> str:
    """Update a running activity when it completes or fails.
    Args:
        agent_name: Name of the agent
        task: Must match the task string used in log_activity
        end_time: ISO 8601 format
        status: 'completed' or 'failed'
        notes: Outcome summary or error details
    """
    existing = (supabase.table("agent_activity_logs")
        .select("id, start_time").eq("agent_name", agent_name)
        .eq("task", task).eq("status", "running")
        .order("start_time", desc=True).limit(1).execute())
    if not existing.data:
        return f"No running activity found for {agent_name}: '{task}'"
    record = existing.data[0]
    start = datetime.fromisoformat(record["start_time"].replace("Z", "+00:00"))
    end = datetime.fromisoformat(end_time.replace("Z", "+00:00"))
    duration = int((end - start).total_seconds())
    update_data = {"end_time": end_time, "status": status, "duration_seconds": duration}
    if notes:
        update_data["notes"] = notes
    supabase.table("agent_activity_logs").update(update_data).eq("id", record["id"]).execute()
    return f"Updated {agent_name} '{task}' → {status} ({duration}s)"


@mcp.tool()
def get_agent_timeline(agent_name: str, date: str) -> str:
    """Get the full activity timeline for a specific agent on a given date.
    Args:
        agent_name: Name of the agent
        date: Date in YYYY-MM-DD format
    """
    result = (supabase.table("agent_activity_logs").select("*")
        .eq("agent_name", agent_name)
        .gte("start_time", f"{date}T00:00:00Z")
        .lte("start_time", f"{date}T23:59:59Z")
        .order("start_time").execute())
    if not result.data:
        return f"No activity for {agent_name} on {date}."
    lines = [f"Timeline for {agent_name} on {date}:\n"]
    for e in result.data:
        dur = f"{e['duration_seconds']}s" if e.get("duration_seconds") else "still running"
        lines.append(f"  [{e['status'].upper()}] {e['task']} ({dur})")
        if e.get("notes"):
            lines.append(f"    → {e['notes']}")
    return "\n".join(lines)


@mcp.tool()
def get_all_agents_summary(date: str) -> str:
    """Get a summary of all agent activity for a given date.
    Args:
        date: Date in YYYY-MM-DD format (e.g. '2026-02-18')
    """
    result = (supabase.table("agent_activity_logs")
        .select("agent_name, task, status, duration_seconds, start_time")
        .gte("start_time", f"{date}T00:00:00Z")
        .lte("start_time", f"{date}T23:59:59Z")
        .order("agent_name, start_time").execute())
    if not result.data:
        return f"No agent activity on {date}."
    by_agent: dict = {}
    for e in result.data:
        by_agent.setdefault(e["agent_name"], []).append(e)
    lines = [f"Agent Activity Summary — {date}\n"]
    for agent, entries in by_agent.items():
        total = sum(e["duration_seconds"] or 0 for e in entries)
        completed = sum(1 for e in entries if e["status"] == "completed")
        failed = sum(1 for e in entries if e["status"] == "failed")
        lines.append(f"\n{agent.upper()} ({len(entries)} tasks, {total}s total)")
        lines.append(f"  Completed: {completed} | Failed: {failed}")
        for e in entries:
            dur = f"{e['duration_seconds']}s" if e.get("duration_seconds") else "running"
            lines.append(f"  • [{e['status']}] {e['task']} ({dur})")
    return "\n".join(lines)


@mcp.tool()
def get_active_agents() -> str:
    """Get all agents currently running (have open/unfinished tasks)."""
    result = (supabase.table("agent_activity_logs")
        .select("agent_name, task, start_time")
        .eq("status", "running").order("start_time").execute())
    if not result.data:
        return "No agents are currently running."
    lines = ["Currently active agents:\n"]
    for e in result.data:
        lines.append(f"  {e['agent_name']}: {e['task']} (started {e['start_time']})")
    return "\n".join(lines)


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

Connecting to Claude Code

Bash

claude mcp add --transport stdio pathfinder \
  -- uv run pathfinder_mcp.py

claude mcp list   # verify
/mcp              # check status inside Claude Code

Making Agents Use It

Add to each subagent's frontmatter:

YAML

---
name: researcher
mcpServers:
  - pathfinder
skills:
  - activity-logging
---

activity-logging skill (preloaded into all agents):

YAML

---
name: activity-logging
user-invocable: false
---

## Activity Logging Protocol

You have access to the Pathfinder MCP server. You MUST log your activity.

START of every task → call `log_activity`:
- agent_name: your name (e.g. "researcher")
- task: brief description
- start_time: current ISO 8601 time (run `date -u +"%Y-%m-%dT%H:%M:%SZ"` to get it)
- status: "running"

END of every task → call `update_activity_status`:
- status: "completed" or "failed"
- end_time: current ISO 8601 time
- notes: 1-2 sentence outcome summary

Do not skip this. Every task must have a log entry.

The Result

Your Pathfinder dashboard shows horizontal bars per agent - a visual timeline of your AI workforce. You can ask Claude:

"Which of my agents has been most efficient this week?"
"Get the researcher's timeline for today"
"Are any agents currently running?"

Because the data is in Supabase, you can query it, visualize it, and feed patterns back to Claude for continuous improvement.

Key Gotchas

No built-in clock: Agents need to run date -u +"%Y-%m-%dT%H:%M:%SZ" or you add a get_current_time() tool to the server
Never print to stdout in a stdio server
Test locally first: uv run pathfinder_mcp.py should start without errors before connecting to Claude Code

Sources

MCP Quickstart - Build a Server (official docs)
MCP Python SDK