Peanut Sheller Server

Source: examples/mcp/peanut_sheller_server.py

Introduction

Run an end-to-end MCP roundtrip for the peanut shelling design brief.

Technical Implementation

This page is generated from the top-of-file module docstring and the example source code. The full script is included below for direct inspection.

  1from __future__ import annotations
  2
  3import argparse
  4import asyncio
  5import os
  6import sys
  7from pathlib import Path
  8from typing import TYPE_CHECKING
  9
 10import design_research_problems as derp
 11
 12if TYPE_CHECKING:
 13    from mcp.server.fastmcp import FastMCP
 14    from mcp.types import CallToolResult
 15
 16PROBLEM_ID = "ideation_peanut_shelling_fu_cagan_kotovsky_2010"
 17SERVER_NAME = "peanut-sheller-demo"
 18
 19
 20def create_server() -> FastMCP:
 21    """Create the MCP server instance for this demo.
 22
 23    Returns:
 24        Configured FastMCP server for the peanut shelling text problem.
 25
 26    Raises:
 27        MissingOptionalDependencyError: If the optional MCP dependency is not
 28            installed.
 29    """
 30    problem = derp.get_problem(PROBLEM_ID)
 31    return problem.to_mcp_server(server_name=SERVER_NAME)
 32
 33
 34def serve_stdio() -> None:
 35    """Run the demo MCP server on stdio transport.
 36
 37    Returns:
 38        None.
 39    """
 40    server = create_server()
 41    server.run(transport="stdio")
 42
 43
 44def _extract_answer(result: CallToolResult) -> str:
 45    """Extract one human-readable answer string from a tool response.
 46
 47    Args:
 48        result: Tool response from ``submit_final``.
 49
 50    Returns:
 51        Submitted answer text when available, or fallback text.
 52    """
 53    structured_content = result.structuredContent
 54    if isinstance(structured_content, dict):
 55        answer = structured_content.get("answer")
 56        if isinstance(answer, str):
 57            return answer
 58
 59    for item in result.content:
 60        text = getattr(item, "text", None)
 61        if isinstance(text, str) and text.strip():
 62            return text.strip()
 63    return "<answer not available in tool payload>"
 64
 65
 66async def run_roundtrip() -> dict[str, object]:
 67    """Start a subprocess MCP server and call its tools.
 68
 69    Returns:
 70        Roundtrip details containing discovered tools and final answer payload.
 71
 72    Raises:
 73        ModuleNotFoundError: If MCP client modules are unavailable.
 74        MissingOptionalDependencyError: If server creation fails due to missing
 75            optional dependency.
 76    """
 77    from mcp.client.session import ClientSession
 78    from mcp.client.stdio import StdioServerParameters, stdio_client
 79
 80    create_server()
 81
 82    example_path = Path(__file__).resolve()
 83    repo_root = example_path.parents[2]
 84    src_path = repo_root / "src"
 85
 86    env = dict(os.environ)
 87    current_pythonpath = env.get("PYTHONPATH")
 88    if current_pythonpath:
 89        env["PYTHONPATH"] = f"{src_path}{os.pathsep}{current_pythonpath}"
 90    else:
 91        env["PYTHONPATH"] = str(src_path)
 92
 93    server_parameters = StdioServerParameters(
 94        command=sys.executable,
 95        args=[str(example_path), "--serve"],
 96        cwd=str(repo_root),
 97        env=env,
 98    )
 99
100    async with stdio_client(server_parameters) as streams:
101        read_stream, write_stream = streams
102        async with ClientSession(read_stream, write_stream) as session:
103            await session.initialize()
104            tools_result = await session.list_tools()
105            submitted_answer = "A hand-cranked sheller with adjustable rollers and manual sorting."
106            call_result = await session.call_tool("submit_final", {"answer": submitted_answer})
107
108    tool_names = sorted(tool.name for tool in tools_result.tools)
109    return {
110        "problem_id": PROBLEM_ID,
111        "tool_count": len(tool_names),
112        "tool_names": tool_names,
113        "answer": _extract_answer(call_result),
114    }
115
116
117def main(argv: list[str] | None = None) -> int:
118    """Run the end-to-end roundtrip or server-only mode.
119
120    Args:
121        argv: Optional CLI argument override.
122
123    Returns:
124        Process exit code.
125    """
126    parser = argparse.ArgumentParser()
127    parser.add_argument(
128        "--serve",
129        action="store_true",
130        help="Run server-only mode for stdio MCP transport.",
131    )
132    args = parser.parse_args(argv)
133
134    if args.serve:
135        try:
136            serve_stdio()
137        except (derp.MissingOptionalDependencyError, ModuleNotFoundError) as exc:
138            print(exc)
139            print("Install the optional MCP dependency with: pip install design-research-problems[mcp]")
140        return 0
141
142    try:
143        summary = asyncio.run(run_roundtrip())
144    except (derp.MissingOptionalDependencyError, ModuleNotFoundError) as exc:
145        print(exc)
146        print("Install the optional MCP dependency with: pip install design-research-problems[mcp]")
147        return 0
148
149    print("Problem id:", summary["problem_id"])
150    print("Tool count:", summary["tool_count"])
151    print("Tools:", ", ".join(str(name) for name in summary["tool_names"]))
152    print("submit_final answer:", summary["answer"])
153    return 0
154
155
156if __name__ == "__main__":
157    raise SystemExit(main())

Expected Results

Run Command

PYTHONPATH=src python3 examples/mcp/peanut_sheller_server.py

Run the command shown below from repository root. Output should summarize the problem setup, a baseline solution, or diagnostic values relevant to this example.