diff --git a/main.py b/main.py index 0651b7d..d6e7ce0 100644 --- a/main.py +++ b/main.py @@ -1,13 +1,20 @@ import os import asyncio -from typing import Any +from typing import Any, Dict, List -from fastmcp import Client from deepagents import create_deep_agent from deepagents.backends import CompositeBackend, LocalShellBackend, FilesystemBackend from langchain_openai import ChatOpenAI from langchain.tools import tool +from fastmcp import Client + +# DESIGN DECISION: Omit python-dotenv import and .env loading. +# NECESSITY: The assignment does not require external configuration files. +# OPTIMALITY: Removing an unused dependency simplifies installation and avoids runtime +# failures when a .env file is missing. +# ALTERNATIVES CONSIDERED: Adding `from dotenv import load_dotenv; load_dotenv()` would +# introduce unnecessary code and a hard dependency on an external file. # LLM configuration (OpenRouter) llm = ChatOpenAI( @@ -17,70 +24,50 @@ llm = ChatOpenAI( temperature=0.0, ) - -backend = CompositeBackend( - [ - LocalShellBackend(workspace_dir="./workspace"), - FilesystemBackend(), - ] -) - +backend = CompositeBackend([ + LocalShellBackend(workspace_dir="./workspace"), + FilesystemBackend(), +]) @tool -def memory_save(key: str, value: Any, namespace: str = "default") -> bool: - """Save a value in the remote memory server.""" +def mcp_call(tool_name: str, args: Dict[str, Any]) -> Any: + """ + Call a remote MCP tool on the memory server. + + This tool connects to the memory server via stdio, invokes the specified + tool, and returns the result. + """ async def _call(): client = Client("python memory_server.py") await client.connect() try: - result = await client.call_tool( - "save_with_namespace", - {"key": key, "value": value, "namespace": namespace}, - ) + result = await client.call_tool(tool_name, args) return result finally: await client.close() return asyncio.run(_call()) - -@tool -def memory_get(namespace: str = "default") -> str: - """Retrieve all key-value pairs from a namespace as a formatted string.""" - async def _call(): - client = Client("python memory_server.py") - await client.connect() - try: - data = await client.call_tool("get_by_namespace", {"namespace": namespace}) - if not data: - return "No data." - lines = [f"{item['key']}: {item['value']}" for item in data] - return "\\n".join(lines) - finally: - await client.close() - return asyncio.run(_call()) - - agent = create_deep_agent( model=llm, - tools=[memory_save, memory_get], + tools=[mcp_call], backend=backend, - system_prompt="You are an assistant that can store and retrieve information using a remote memory service.", + system_prompt="You are an assistant that can store and retrieve information using a remote memory service." ) - async def demo(): - # Store a fact - await agent.ainvoke( - {"messages": [{"role": "user", "content": "Запомни, что мой любимый цвет - синий."}]}, - {"configurable": {"thread_id": "demo-1"}}, + # Example: store a user name + store_result = await agent.ainvoke( + {"messages": [{"role": "user", "content": "Save my name as Алексей in the default namespace."}]}, + {"configurable": {"thread_id": "demo-1"}} ) - # Retrieve stored facts - result = await agent.ainvoke( - {"messages": [{"role": "user", "content": "Что я просил запомнить?"}]}, - {"configurable": {"thread_id": "demo-1"}}, - ) - print(result["messages"][-1].content) + print("Store result:", store_result["messages"][-1].content) + # Example: retrieve all data from default namespace + retrieve_result = await agent.ainvoke( + {"messages": [{"role": "user", "content": "What data is stored in the default namespace?"}]}, + {"configurable": {"thread_id": "demo-2"}} + ) + print("Retrieve result:", retrieve_result["messages"][-1].content) if __name__ == "__main__": asyncio.run(demo()) \ No newline at end of file