Last active
October 21, 2025 01:34
-
-
Save lukestanley/d9819f8f3f387d62dace9e09dd58187b to your computer and use it in GitHub Desktop.
Revisions
-
lukestanley revised this gist
Oct 21, 2025 . 1 changed file with 36 additions and 36 deletions.There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode charactersOriginal file line number Diff line number Diff line change @@ -23,22 +23,51 @@ from mcp import ClientSession, StdioServerParameters from mcp.client.sse import sse_client from mcp.client.stdio import stdio_client from icecream import ic from keys import OPENROUTER_API_KEY LLM=dspy.LM("openrouter/openai/gpt-5-nano", api_key=OPENROUTER_API_KEY, temperature=1.0, max_tokens = 16000) # Easily set multiple kinds of MCP servers: SERVERS = [ "npx -y @modelcontextprotocol/server-memory", "https://mcp.deepwiki.com/sse", "uvx mcp-server-fetch", "npx -y @modelcontextprotocol/server-sequential-thinking", "npx -y @playwright/mcp@latest", ] class Ask(dspy.Signature): question: str = dspy.InputField() answer: str = dspy.OutputField() source = open(__file__).read() # A demo question: q=f"""Script: <file> {source} </file> 1. Explain what the script does in plain English with emojis using up to 30 words. 2. Summarise your tools with raw tool names, e.g: read_wiki_structure first. 3. Summarise the top story that is on BOTH YC News and BBC News. """ ic(q) # Lists the MCP server tools # A more simple API interface: async def main(): dspy.configure(lm=LLM, callbacks=[DSPyLoggingCallback()]) async with get_mcp_tool_session(SERVERS) as (tools, call_tool): agent = dspy.ReAct(Ask, tools=tools) output = await agent.acall( question=q ) print(output.answer) # The wrapper: @asynccontextmanager async def get_mcp_tool_session(servers: list[str]): async with AsyncExitStack() as stack: @@ -70,34 +99,5 @@ def on_module_end(self, call_id, outputs, exception): print(f" {k}: {v}") print("") if __name__ == "__main__": asyncio.run(main()) -
Oct 21, 2025 .There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode charactersOriginal file line number Diff line number Diff line change @@ -0,0 +1,103 @@ #!/usr/bin/env -S uv run # Uses the inline script metadata Python Enhancment Proposal (PEP 723) # To install UV see: https://docs.astral.sh/uv/getting-started/installation/ # To run with UV: uv run dspy_minimal_mcp_demo.py # /// script # requires-python = ">=3.11" # dependencies = [ # "dspy>=3.0.2", # "mcp>=1.2.0", # "litellm>=1.76.0", # "icecream" # ] # /// import asyncio import shlex from contextlib import asynccontextmanager, AsyncExitStack import dspy from dspy import Tool from dspy.utils.callback import BaseCallback from mcp import ClientSession, StdioServerParameters from mcp.client.sse import sse_client from mcp.client.stdio import stdio_client from icecream import ic from keys import OPENROUTER_API_KEY LLM=dspy.LM("openrouter/openai/gpt-5-mini", api_key=OPENROUTER_API_KEY, temperature=1.0, max_tokens = 16000) # Multiple kinds of MCP servers easily added here: SERVERS = [ # "npx -y @modelcontextprotocol/server-memory", "https://mcp.deepwiki.com/sse", # "uvx mcp-server-fetch", # "npx -y @modelcontextprotocol/server-sequential-thinking", "npx -y @playwright/mcp@latest", ] @asynccontextmanager async def get_mcp_tool_session(servers: list[str]): async with AsyncExitStack() as stack: all_tools: list[Tool] = [] for spec in servers: if spec.startswith("http://") or spec.startswith("https://"): read, write = await stack.enter_async_context(sse_client(url=spec)) else: parts = shlex.split(spec) params = StdioServerParameters(command=parts[0], args=parts[1:], env=None) read, write = await stack.enter_async_context(stdio_client(params)) session = await stack.enter_async_context(ClientSession(read, write)) await session.initialize() tool_list = await session.list_tools() all_tools.extend(Tool.from_mcp_tool(session, desc) for desc in tool_list.tools) async def call_tool(name: str, **kwargs): tool = next(t for t in all_tools if t.name == name) return await tool.acall(**kwargs) yield all_tools, call_tool # Just for logging: class DSPyLoggingCallback(BaseCallback): def on_module_end(self, call_id, outputs, exception): for k, v in outputs.items(): print(f" {k}: {v}") print("") # Currently, a raw and unoptimised signature: class Ask(dspy.Signature): question: str = dspy.InputField() answer: str = dspy.OutputField() source = open(__file__).read() q=f"""Script: <file> {source} </file> 1. Explain what the script does in plain English with emojis using up to 30 words. 2. Summarise your tools with raw tool names, e.g: read_wiki_structure first. 3. Summarise the top story that is on BOTH YC News and BBC News. """ ic(q) # Lists the MCP server tools # A more simple API interface: async def main(): dspy.configure(lm=LLM, callbacks=[DSPyLoggingCallback()]) async with get_mcp_tool_session(SERVERS) as (tools, call_tool): agent = dspy.ReAct(Ask, tools=tools) output = await agent.acall( question=q ) print(output.answer) if __name__ == "__main__": asyncio.run(main())