MCP Use Framework
mcp-use is a full-stack Python framework for building Multi-Cloud Platform (MCP) agents, clients, and servers. It simplifies the development of distributed systems with a focus on asynchronous communication. The current version is 1.7.0 and it maintains an active release cadence, frequently adding features and improvements.
Warnings
- breaking mcp-use requires Python 3.11 or newer. Attempting to install or run on older Python versions (e.g., 3.10, 3.9) will result in syntax errors, dependency resolution failures, or runtime exceptions due to modern language features and type hints used throughout the library.
- gotcha All core operations in mcp-use (e.g., `Agent.start()`, `Client.connect()`, `Channel.send()`) are `async` functions and must be `await`ed within an `asyncio` event loop. Forgetting `await` or trying to call them synchronously will lead to `RuntimeWarning: coroutine '...' was never awaited` or `TypeError`.
- gotcha Improper resource cleanup (e.g., not disconnecting clients, stopping agents, or closing channels) can lead to resource leaks, open network connections, and processes that don't terminate cleanly. This is particularly critical in long-running services or tests.
- gotcha Error handling across distributed components (Agent, Client, Server) requires careful consideration. Uncaught exceptions on one side might silently fail or lead to connection drops without clear indication on the other.
Install
-
pip install mcp-use
Imports
- Agent
from mcp_use import Agent
- Client
from mcp_use import Client
- Server
from mcp_use import Server
- Channel
from mcp_use import Channel
Quickstart
import asyncio
from mcp_use import Agent, Client, Server, Channel
async def main():
# Example: A simple MCP Agent and Client interaction
# 1. Start a simple Agent
class MyAgent(Agent):
async def handle_message(self, message: str) -> str:
print(f"Agent received: {message}")
return f"Agent processed: {message.upper()}"
agent = MyAgent(name="test-agent", host="127.0.0.1", port=8000)
agent_task = asyncio.create_task(agent.start())
await asyncio.sleep(0.5) # Give agent time to start
# 2. Connect a Client to the Agent
client = Client(name="test-client", host="127.0.0.1", port=8000)
await client.connect()
# 3. Send a message via a Channel
try:
async with client.channel("my-channel") as channel:
response = await channel.send("hello mcp!")
print(f"Client received: {response}")
assert response == "Agent processed: HELLO MCP!"
finally:
# 4. Clean up
await client.disconnect()
agent_task.cancel()
await agent.stop()
print("Cleanup complete.")
if __name__ == "__main__":
asyncio.run(main())