73 lines
2.4 KiB
Python
73 lines
2.4 KiB
Python
import os
|
|
import asyncio
|
|
from typing import Any, Dict, List
|
|
|
|
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(
|
|
model="openai/gpt-oss-20b:free",
|
|
base_url="https://openrouter.ai/api/v1",
|
|
api_key=os.getenv("OPENAI_API_KEY"),
|
|
temperature=0.0,
|
|
)
|
|
|
|
backend = CompositeBackend([
|
|
LocalShellBackend(workspace_dir="./workspace"),
|
|
FilesystemBackend(),
|
|
])
|
|
|
|
@tool
|
|
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(tool_name, args)
|
|
return result
|
|
finally:
|
|
await client.close()
|
|
return asyncio.run(_call())
|
|
|
|
agent = create_deep_agent(
|
|
model=llm,
|
|
tools=[mcp_call],
|
|
backend=backend,
|
|
system_prompt="You are an assistant that can store and retrieve information using a remote memory service."
|
|
)
|
|
|
|
async def demo():
|
|
# 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"}}
|
|
)
|
|
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()) |