mcp-categorical

MCP Categorical Tool Composition

Categorical patterns for Model Context Protocol server design and tool composition.

Core Categorical Mapping

MCP structures map to category theory:

• Tools: Morphisms Parameters → Result
• Resources: Objects with URI-based identity
• Prompts: Exponential objects Context → Prompt
• Context: Product type of available information
• Server: Category of tools with composition

Basic MCP Server (TypeScript)

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";

// Create server (category of tools)
const server = new Server(
  { name: "categorical-server", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

// Define tool (morphism)
server.setRequestHandler(ListToolsRequestSchema, async () => ({
  tools: [
    {
      name: "analyze",
      description: "Analyze input with categorical structure",
      inputSchema: {
        type: "object",
        properties: {
          input: { type: "string" },
          category: { type: "string" }
        },
        required: ["input"]
      }
    }
  ]
}));

// Tool implementation (morphism execution)
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;
  
  if (name === "analyze") {
    return {
      content: [{ type: "text", text: `Analyzed: ${args.input}` }]
    };
  }
  
  throw new Error(`Unknown tool: ${name}`);
});

// Run server
const transport = new StdioServerTransport();
server.connect(transport);

Categorical Tool Composition

Tool as Functor

import { z } from "zod";

// Tool type: Parameters → Effect<Result, Error>
interface Tool<P, R> {
  name: string;
  schema: z.ZodType<P>;
  execute: (params: P) => Promise<R>;
}

// Functor: map over tool results
function mapTool<P, R, S>(
  tool: Tool<P, R>,
  f: (r: R) => S
): Tool<P, S> {
  return {
    name: tool.name,
    schema: tool.schema,
    execute: async (params) => f(await tool.execute(params))
  };
}

// Example
const searchTool: Tool<{ query: string }, string[]> = {
  name: "search",
  schema: z.object({ query: z.string() }),
  execute: async ({ query }) => [`Result for: ${query}`]
};

const formattedSearch = mapTool(searchTool, (results) =>
  results.join("\n")
);

Tool Composition (Kleisli)

// Compose tools: (A → B) → (B → C) → (A → C)
function composeTool<A, B, C>(
  first: Tool<A, B>,
  second: Tool<B, C>
): Tool<A, C> {
  return {
    name: `${first.name}_then_${second.name}`,
    schema: first.schema,
    execute: async (params: A) => {
      const intermediate = await first.execute(params);
      return second.execute(intermediate as any);
    }
  };
}

// Product of tools: run in parallel
function productTool<A, B, C>(
  tool1: Tool<A, B>,
  tool2: Tool<A, C>
): Tool<A, [B, C]> {
  return {
    name: `${tool1.name}_and_${tool2.name}`,
    schema: tool1.schema,
    execute: async (params: A) => {
      const [r1, r2] = await Promise.all([
        tool1.execute(params),
        tool2.execute(params)
      ]);
      return [r1, r2];
    }
  };
}

Natural Transformation Between Tool Categories

// Transform tool category to another
interface ToolTransform<F, G> {
  transform: <P, R>(tool: Tool<P, R>) => Tool<P, R>;
}

// Logging transformation
const withLogging: ToolTransform<any, any> = {
  transform: (tool) => ({
    ...tool,
    execute: async (params) => {
      console.log(`Calling ${tool.name} with`, params);
      const result = await tool.execute(params);
      console.log(`Result from ${tool.name}:`, result);
      return result;
    }
  })
};

// Retry transformation
const withRetry = (maxRetries: number): ToolTransform<any, any> => ({
  transform: (tool) => ({
    ...tool,
    execute: async (params) => {
      for (let i = 0; i < maxRetries; i++) {
        try {
          return await tool.execute(params);
        } catch (e) {
          if (i === maxRetries - 1) throw e;
        }
      }
      throw new Error("Unreachable");
    }
  })
});

Python MCP Server

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

# Categorical tool registry
class ToolCategory:
    """Category of tools with composition."""
    
    def __init__(self):
        self.tools: dict[str, Tool] = {}
        self.handlers: dict[str, callable] = {}
    
    def register(self, name: str, schema: dict, handler: callable):
        """Register tool as morphism."""
        self.tools[name] = Tool(
            name=name,
            description=f"Tool: {name}",
            inputSchema=schema
        )
        self.handlers[name] = handler
    
    def compose(self, name1: str, name2: str) -> str:
        """Compose two tools."""
        composed_name = f"{name1}_then_{name2}"
        
        async def composed_handler(params):
            result1 = await self.handlers[name1](params)
            return await self.handlers[name2](result1)
        
        self.handlers[composed_name] = composed_handler
        return composed_name

# Create server
app = Server("categorical-mcp")
category = ToolCategory()

# Register tools
category.register(
    "fetch",
    {"type": "object", "properties": {"url": {"type": "string"}}},
    lambda p: f"Fetched: {p['url']}"
)

category.register(
    "parse",
    {"type": "object", "properties": {"content": {"type": "string"}}},
    lambda p: f"Parsed: {p['content']}"
)

@app.list_tools()
async def list_tools():
    return list(category.tools.values())

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    handler = category.handlers.get(name)
    if handler:
        result = await handler(arguments) if asyncio.iscoroutinefunction(handler) else handler(arguments)
        return [TextContent(type="text", text=str(result))]
    raise ValueError(f"Unknown tool: {name}")

# Run
async def main():
    async with stdio_server() as (read, write):
        await app.run(read, write)

Resource Management (Objects)

// Resources as objects in category
interface Resource<T> {
  uri: string;
  name: string;
  mimeType: string;
  read: () => Promise<T>;
}

// Resource functor
function mapResource<T, U>(
  resource: Resource<T>,
  f: (t: T) => U
): Resource<U> {
  return {
    ...resource,
    read: async () => f(await resource.read())
  };
}

// Resource registry
class ResourceCategory {
  private resources: Map<string, Resource<any>> = new Map();
  
  register<T>(resource: Resource<T>): void {
    this.resources.set(resource.uri, resource);
  }
  
  get<T>(uri: string): Resource<T> | undefined {
    return this.resources.get(uri);
  }
  
  // Natural transformation: Resource → Tool
  toTool<T>(uri: string): Tool<{}, T> {
    const resource = this.get<T>(uri);
    if (!resource) throw new Error(`Unknown resource: ${uri}`);
    
    return {
      name: `read_${resource.name}`,
      schema: z.object({}),
      execute: async () => resource.read()
    };
  }
}

Typed Context Protocol

import { z } from "zod";

// Context as product type
const ContextSchema = z.object({
  conversation: z.array(z.object({
    role: z.enum(["user", "assistant"]),
    content: z.string()
  })),
  tools: z.array(z.string()),
  resources: z.array(z.string()),
  metadata: z.record(z.unknown())
});

type Context = z.infer<typeof ContextSchema>;

// Context functor operations
const contextOps = {
  // Add message (endofunctor)
  addMessage: (ctx: Context, role: "user" | "assistant", content: string): Context => ({
    ...ctx,
    conversation: [...ctx.conversation, { role, content }]
  }),
  
  // Filter tools (subfunctor)
  filterTools: (ctx: Context, predicate: (t: string) => boolean): Context => ({
    ...ctx,
    tools: ctx.tools.filter(predicate)
  }),
  
  // Merge contexts (coproduct)
  merge: (ctx1: Context, ctx2: Context): Context => ({
    conversation: [...ctx1.conversation, ...ctx2.conversation],
    tools: [...new Set([...ctx1.tools, ...ctx2.tools])],
    resources: [...new Set([...ctx1.resources, ...ctx2.resources])],
    metadata: { ...ctx1.metadata, ...ctx2.metadata }
  })
};

Prompt Templates (Exponential Objects)

// Prompt as exponential: Context → String
interface PromptTemplate {
  name: string;
  template: (ctx: Context) => string;
}

// Prompt composition
function composePrompts(
  p1: PromptTemplate,
  p2: PromptTemplate
): PromptTemplate {
  return {
    name: `${p1.name}_then_${p2.name}`,
    template: (ctx) => `${p1.template(ctx)}\n\n${p2.template(ctx)}`
  };
}

// Example prompts
const systemPrompt: PromptTemplate = {
  name: "system",
  template: (ctx) => `You have access to: ${ctx.tools.join(", ")}`
};

const contextPrompt: PromptTemplate = {
  name: "context",
  template: (ctx) => ctx.conversation
    .map(m => `${m.role}: ${m.content}`)
    .join("\n")
};

Categorical Guarantees

MCP with categorical patterns ensures:

1. Tool Composability: Tools compose via Kleisli composition
2. Type Safety: Zod schemas validate parameters at boundaries
3. Resource Identity: URIs provide unique object identification
4. Context Coherence: Product structure preserves all context
5. Natural Transformations: Tool transformations preserve structure