Make MCP+Agents part of lynxkite-lynxscribe.

examples/LynxScribe/MCP/demo.py

@@ -1,15 +1,5 @@
-import enum
 from lynxkite_core import ops
 from lynxkite_graph_analytics import core
-import fastapi
-from lynxscribe.core.llm.base import get_llm_engine
-import contextlib
-import mcp
-from mcp.client.stdio import stdio_client
-from lynxscribe.components.task_solver import TaskSolver
-from lynxscribe.core.models.prompts import Function, Tool
-
-FRONTEND_URL = "http://localhost:8501/"
 
 lsop = ops.op_registration(
     "LynxKite Graph Analytics", "LynxScribe", dir="bottom-to-top", color="blue"
@@ -17,47 +7,6 @@ lsop = ops.op_registration(
 dsop = ops.op_registration("LynxKite Graph Analytics", "Data Science")
 
 
-@lsop("Chat frontend", color="gray", outputs=[], view="service")
-def chat_frontend(agent: dict):
-    agent = agent["agent"]
-    return Agent(
-        agent["name"],
-        agent["description"],
-        agent["system_prompt"],
-        agent["mcp_servers"],
-        [chat_frontend(a) for a in agent["sub_agents"]],
-    )
-
-
-@lsop("Agent")
-def agent(
-    tools: list[dict],
-    *,
-    name: str = "",
-    description: ops.LongStr = "This agent helps with various tasks.",
-    system_prompt: ops.LongStr = "You are a helpful assistant.",
-):
-    prompt = [system_prompt]
-    for tool in tools:
-        if tool.get("extra_prompt"):
-            prompt.append(tool["extra_prompt"])
-    return {
-        "agent": {
-            "name": name,
-            "description": description,
-            "system_prompt": "\n".join(prompt),
-            "mcp_servers": [t["command"] for t in tools if "command" in t],
-            "sub_agents": [t for t in tools if "agent" in t],
-        }
-    }
-
-
-@lsop("MCP: Custom", color="green")
-def custom_mcp(*, command: str, extra_prompt: ops.LongStr):
-    if command.strip():
-        return {"command": command.strip().split(), "extra_prompt": extra_prompt}
-
-
 @lsop("MCP: Query database with SQL", color="green")
 def sql_tool(db: str):
     return {
@@ -75,189 +24,3 @@ is out of order currently. Please use `query` instead."
 def db(data_pipeline: list[core.Bundle]):
     # The source of this file: https://github.com/biggraph/lynxscribe/pull/416
     return "movie_data.sqlite.db"
-
-
-class MCPSearchEngine(str, enum.Enum):
-    Google = "Google"
-    Bing = "Bing"
-    DuckDuckGo = "DuckDuckGo"
-
-
-@lsop("MCP: Search web", color="green")
-def web_search(*, engine: MCPSearchEngine = MCPSearchEngine.Google):
-    match engine:
-        case MCPSearchEngine.Google:
-            return {"command": ["npx", "-y", "https://github.com/pskill9/web-search"]}
-        case MCPSearchEngine.Bing:
-            return {"command": ["uvx", "bing-search-mcp"]}
-        case MCPSearchEngine.DuckDuckGo:
-            return {"command": ["uvx", "duckduckgo-mcp-server"]}
-
-
-@lsop("MCP: Financial data", color="green")
-def financial_data():
-    return {"command": ["uvx", "investor-agent"]}
-
-
-@lsop("MCP: Calculator", color="green")
-def calculator():
-    return {"command": ["uvx", "mcp-server-calculator"]}
-
-
-class Agent:
-    def __init__(
-        self,
-        name: str,
-        description: str,
-        prompt: str,
-        mcp_servers: list[list[str]],
-        agents: list["Agent"],
-    ):
-        self.name = name
-        self.description = description
-        self.prompt = prompt
-        self.mcp_servers = mcp_servers
-        self.agents = agents
-        self.mcp_client = None
-        self.task_solver = None
-
-    async def init(self):
-        if self.task_solver is not None:
-            return
-        self.mcp_client = MultiMCPClient()
-        await self.mcp_client.connect(self.mcp_servers)
-        agents_as_functions = [agent.as_function() for agent in self.agents]
-        self.task_solver = TaskSolver(
-            llm=get_llm_engine(name="openai"),
-            model="gpt-4.1-nano",
-            initial_messages=[self.prompt],
-            functions=[*self.mcp_client.tools, *agents_as_functions],
-            tool_choice="required",
-            temperature=0.0,
-            max_tool_call_steps=999,
-        )
-
-    def get_description(self, url: str) -> str:
-        return f"[Go to frontend]({FRONTEND_URL}?service={url}/chat/completions)"
-
-    async def get(self, request: fastapi.Request) -> dict:
-        if request.state.remaining_path == "models":
-            return {
-                "object": "list",
-                "data": [
-                    {
-                        "id": "LynxScribe",
-                        "object": "model",
-                        "created": 0,
-                        "owned_by": "lynxkite",
-                        "meta": {"profile_image_url": "https://lynxkite.com/favicon.png"},
-                    }
-                ],
-            }
-        return {"error": "Not found"}
-
-    async def post(self, request: fastapi.Request) -> dict:
-        if request.state.remaining_path == "chat/completions":
-            request = await request.json()
-            assert not request["stream"]
-            await self.init()
-            res = await self.task_solver.solve(request["messages"][-1]["content"])
-            return {"choices": [{"message": {"role": "assistant", "content": res}}]}
-
-        return {"error": "Not found"}
-
-    def as_function(self):
-        """A callable that can be used as a tool by another Agent."""
-
-        # Find the value of x given that 4*x^4 = 44. Compute the numerical value.
-        async def ask(message):
-            print(f"Calling agent {self.name} with message: {message}")
-            await self.init()
-            res = await self.task_solver.solve(message)
-            print(f"Agent {self.name} response: {res}")
-            return res
-
-        ask.__name__ = "ask_" + self.name.lower().replace(" ", "_")
-        ask._tool = Tool(
-            type="function",
-            function=Function(
-                name=ask.__name__,
-                description=self.description,
-                parameters={
-                    "type": "object",
-                    "required": ["message"],
-                    "properties": {
-                        "message": {
-                            "type": "string",
-                            "description": "The question to ask the agent.",
-                        }
-                    },
-                },
-            ),
-        )
-        return ask
-
-
-class MCPClient:
-    def __init__(self):
-        self.session: mcp.ClientSession | None = None
-        self.exit_stack = contextlib.AsyncExitStack()
-
-    async def connect_to_server(self, config):
-        """Connect to an MCP server."""
-        server_params = mcp.StdioServerParameters(**config)
-        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
-        self.stdio, self.write = stdio_transport
-        self.session = await self.exit_stack.enter_async_context(
-            mcp.ClientSession(self.stdio, self.write)
-        )
-        await self.session.initialize()
-        self.connected = True
-        response = await self.session.list_tools()
-        self.tools = [self.to_func(tool) for tool in response.tools]
-
-    async def run_tool(self, tool_name: str, tool_args: dict):
-        print(f"Calling tool {tool_name} with args {tool_args}")
-        return await self.session.call_tool(tool_name, tool_args)
-
-    def to_func(self, tool):
-        """Convert a tool to a callable function."""
-
-        async def func(**kwargs):
-            res = await self.run_tool(tool.name, kwargs)
-            return res.content[0].text
-
-        func.__name__ = tool.name
-        func._tool = Tool(
-            type="function",
-            function=Function(
-                name=tool.name,
-                description=tool.description,
-                parameters=tool.inputSchema,
-            ),
-        )
-        return func
-
-    async def cleanup(self):
-        await self.exit_stack.aclose()
-
-
-class MultiMCPClient:
-    """A client that can connect to multiple MCP servers."""
-
-    async def connect(self, servers: list[str]):
-        self.clients = []
-        for server in servers:
-            client = MCPClient()
-            [command, *args] = server
-            await client.connect_to_server({"command": command, "args": args})
-            self.clients.append(client)
-        tools = []
-        for client in self.clients:
-            tools.extend(client.tools)
-        self.tools = tools
-
-    async def cleanup(self):
-        # Reverse cleanup as in https://github.com/microsoft/semantic-kernel/issues/12627.
-        for client in reversed(self.clients):
-            await client.cleanup()

examples/LynxScribe/MCP/requirements.txt

@@ -1,3 +0,0 @@
-mcp
-streamlit
-../../lynxscribe

lynxkite-lynxscribe/src/lynxkite_lynxscribe/__init__.py

@@ -1,3 +1,4 @@
+from . import agentic  # noqa (imported to trigger registration)
 from . import lynxscribe_ops  # noqa (imported to trigger registration)
 from . import llm_ops  # noqa (imported to trigger registration)
 from .lynxscribe_ops import api_service_post, api_service_get

lynxkite-lynxscribe/src/lynxkite_lynxscribe/agentic.py

@@ -0,0 +1,190 @@
+"""Operations that allow LynxScribe agents and MCP servers to be combined with LynxKite pipelines."""
+
+import enum
+import os
+import typing
+
+from lynxkite_core import ops
+from lynxscribe.components.task_solver import TaskSolver
+from lynxscribe.components.mcp import MCPClient
+from lynxscribe.core.llm.base import get_llm_engine
+from lynxscribe.core.models.prompts import Function, Tool
+
+if typing.TYPE_CHECKING:
+    import fastapi
+
+CHAT_FRONTEND_URL = os.environ.get("CHAT_FRONTEND_URL", "http://localhost:8501/")
+
+op = ops.op_registration(
+    "LynxKite Graph Analytics", "LynxScribe", dir="bottom-to-top", color="blue"
+)
+
+
+@op("Chat frontend", color="gray", outputs=[], view="service")
+def chat_frontend(agent: dict):
+    agent = agent["agent"]
+    return Agent(
+        agent["name"],
+        agent["description"],
+        agent["system_prompt"],
+        agent["mcp_servers"],
+        [chat_frontend(a) for a in agent["sub_agents"]],
+    )
+
+
+@op("Agent")
+def agent(
+    tools: list[dict],
+    *,
+    name: str = "",
+    description: ops.LongStr = "This agent helps with various tasks.",
+    system_prompt: ops.LongStr = "You are a helpful assistant.",
+):
+    prompt = [system_prompt]
+    for tool in tools:
+        if tool.get("extra_prompt"):
+            prompt.append(tool["extra_prompt"])
+    return {
+        "agent": {
+            "name": name,
+            "description": description,
+            "system_prompt": "\n".join(prompt),
+            "mcp_servers": [t["command"] for t in tools if "command" in t],
+            "sub_agents": [t for t in tools if "agent" in t],
+        }
+    }
+
+
+@op("MCP: Custom", color="green")
+def custom_mcp(*, command: str, extra_prompt: ops.LongStr):
+    if command.strip():
+        return {"command": command.strip().split(), "extra_prompt": extra_prompt}
+
+
+# A few basic MCP integrations to serve as examples.
+
+
+class MCPSearchEngine(str, enum.Enum):
+    Google = "Google"
+    Bing = "Bing"
+    DuckDuckGo = "DuckDuckGo"
+
+
+@op("MCP: Search web", color="green")
+def web_search(*, engine: MCPSearchEngine = MCPSearchEngine.Google):
+    match engine:
+        case MCPSearchEngine.Google:
+            return {"command": ["npx", "-y", "https://github.com/pskill9/web-search"]}
+        case MCPSearchEngine.Bing:
+            return {"command": ["uvx", "bing-search-mcp"]}
+        case MCPSearchEngine.DuckDuckGo:
+            return {"command": ["uvx", "duckduckgo-mcp-server"]}
+
+
+@op("MCP: Financial data", color="green")
+def financial_data():
+    return {"command": ["uvx", "investor-agent"]}
+
+
+@op("MCP: Calculator", color="green")
+def calculator():
+    return {"command": ["uvx", "mcp-server-calculator"]}
+
+
+@op("MCP: Time", color="green")
+def time():
+    return {"command": ["uvx", "mcp-server-time"]}
+
+
+class Agent:
+    def __init__(
+        self,
+        name: str,
+        description: str,
+        prompt: str,
+        mcp_servers: list[list[str]],
+        agents: list["Agent"],
+    ):
+        self.name = name
+        self.description = description
+        self.prompt = prompt
+        self.mcp_servers = mcp_servers
+        self.agents = agents
+        self.mcp_client = None
+        self.task_solver = None
+
+    async def init(self):
+        if self.task_solver is not None:
+            return
+        self.mcp_client = MCPClient()
+        await self.mcp_client.connect(self.mcp_servers)
+        agents_as_functions = [agent.as_function() for agent in self.agents]
+        self.task_solver = TaskSolver(
+            llm=get_llm_engine(name="openai"),
+            model="gpt-4.1-nano",
+            initial_messages=[self.prompt],
+            functions=[*self.mcp_client.tools, *agents_as_functions],
+            tool_choice="required",
+            temperature=0.0,
+            max_tool_call_steps=999,
+        )
+
+    def get_description(self, url: str) -> str:
+        return f"[Go to frontend]({CHAT_FRONTEND_URL}?service={url}/chat/completions)"
+
+    async def get(self, request: "fastapi.Request") -> dict:
+        if request.state.remaining_path == "models":
+            return {
+                "object": "list",
+                "data": [
+                    {
+                        "id": "LynxScribe",
+                        "object": "model",
+                        "created": 0,
+                        "owned_by": "lynxkite",
+                        "meta": {"profile_image_url": "https://lynxkite.com/favicon.png"},
+                    }
+                ],
+            }
+        return {"error": "Not found"}
+
+    async def post(self, request: "fastapi.Request") -> dict:
+        if request.state.remaining_path == "chat/completions":
+            request = await request.json()
+            assert not request["stream"]
+            await self.init()
+            res = await self.task_solver.solve(request["messages"][-1]["content"])
+            return {"choices": [{"message": {"role": "assistant", "content": res}}]}
+
+        return {"error": "Not found"}
+
+    def as_function(self):
+        """A callable that can be used as a tool by another Agent."""
+
+        # Find the value of x given that 4*x^4 = 44. Compute the numerical value.
+        async def ask(message):
+            print(f"Calling agent {self.name} with message: {message}")
+            await self.init()
+            res = await self.task_solver.solve(message)
+            print(f"Agent {self.name} response: {res}")
+            return res
+
+        ask.__name__ = "ask_" + self.name.lower().replace(" ", "_")
+        ask._tool = Tool(
+            type="function",
+            function=Function(
+                name=ask.__name__,
+                description=self.description,
+                parameters={
+                    "type": "object",
+                    "required": ["message"],
+                    "properties": {
+                        "message": {
+                            "type": "string",
+                            "description": "The question to ask the agent.",
+                        }
+                    },
+                },
+            ),
+        )
+        return ask