Hi, I am trying to figure out what is the best way to use MCP server in agno workflow.
So far, I’ve managed to get a basic async workflow running using an agent, like this:
from agno.workflow import RunEvent, RunResponse, Workflow
from agno.agent import Agent
from utils import get_model_for_provider
import asyncio
class SimpleWorkflow(Workflow):
sum_agent: Agent = Agent(
model=get_model_for_provider('azureopenai:gpt-4o'),
debug_mode=True,
markdown=True,
show_tool_calls=True,
use_json_mode=True,
)
async def arun(self, message: str) -> RunResponse:
"""Run the workflow with the given message."""
response = await self.sum_agent.arun(message)
return response
async def main():
workflow = SimpleWorkflow()
message = "what is 1+1?"
response = await workflow.arun(message=message)
print(response.content)
if __name__ == "__main__":
asyncio.run(main())
Now I want to wrap the agent logic inside an async with block using MCPTools, like this:
async with MCPTools(command="uvx mcp-server-git") as mcp_tools:
sum_agent: Agent = Agent(...)
The problem is: since sum_agent is a class attribute and needs to be accessed within the arun() method, I’m not sure how to initialize it properly inside a context manager while still keeping it accessible in the workflow.
Question:
How can I correctly integrate the MCPTools context manager with an agent in the Workflow class structure, given that the agent needs to be initialized within the context but also be accessible as a class variable?
Any advice or code examples would be really appreciated. Thanks!
Hi @JSo
thanks for reaching out and supporting Agno. I’ve shared this with the team, we’re working through all requests one by one and will get back to you soon.
If it’s urgent, please let us know. We appreciate your patience!
There are probably multiple ways to achieve this! It’s all about making sure the lifecycle of MCPTools, acting as the context manager, hasn’t ended when the agent needs it. What I would do here is to initialize the Agent inside a function, to keep tighter control of its lifecycle:
import asyncio
from textwrap import dedent
from agno.agent import Agent
from agno.tools.mcp import MCPTools
from agno.workflow import Workflow
class SimpleWorkflow(Workflow):
...
async def arun_workflow_agent(self, message: str):
# Context manager and Agent lifecyles are encapsulated in this function
async with MCPTools(command="uvx mcp-server-git") as mcp_tools:
sum_agent: Agent = Agent(
tools=[mcp_tools],
instructions=dedent("""\
You are a GitHub assistant. Help users explore repositories and their activity.
- Use headings to organize your responses
- Be concise and focus on relevant information\
"""),
debug_mode=True,
markdown=True,
show_tool_calls=True,
use_json_mode=True,
)
await sum_agent.aprint_response(message)
if __name__ == "__main__":
workflow = SimpleWorkflow()
asyncio.run(
workflow.arun_workflow_agent(
message="Tell me about Agno. Github repo: https://github.com/agno-agi/agno"
)
)
Let me know if this works for you. Else I’m happy to keep helping!
Hi @manu thanks for the response and the suggestion. I think this is a way to initialise agent with MCP tool but this can be quite inefficient when multiple nodes in the workflow need to reuse the agent as everytime it execute, it initialise agent fresh and initialise mcp servers as well. ideally the agent is initialised with MCP tool and throughout the workflow it can be called whenever and whereever. is there a way to make it work in that way?
Hey @JSo, you are right. I don’t think there’s a simple way to build this right now with Workflows. You could take a look at how teams using MCP tools look like, maybe that pattern works for what you are trying to achieve. For example:
We will soon work on offering a better way to initialize the MCPTools (without it being an async context manager) for setups like this one to become easier to handle.