mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-01-11 09:08:31 +00:00
a270742319
* mcp server proposal * Refactor MCP server implementation: rename MCPServer to MCPServerAdapter and update usage examples. Adjust error message for optional dependencies installation. * Update MCPServerAdapter usage examples to remove unnecessary parameters in context manager instantiation. * Refactor MCPServerAdapter to move optional dependency imports inside the class constructor, improving error handling for missing dependencies. * Enhance MCPServerAdapter by adding type hinting for server parameters and improving error handling during server startup. Optional dependency imports are now conditionally loaded, ensuring clearer error messages for missing packages. * Refactor MCPServerAdapter to improve error handling for missing 'mcp' package. Conditional imports are now used, prompting users to install the package if not found, enhancing user experience during server initialization. * Refactor MCPServerAdapter to ensure proper cleanup after usage. Removed redundant exception handling and ensured that the server stops in a finally block, improving resource management. * add documentation * fix typo close -> stop * add tests and fix double call with context manager * Enhance MCPServerAdapter with logging capabilities and improved error handling during initialization. Added logging for cleanup errors and refined the structure for handling missing 'mcp' package dependencies. --------- Co-authored-by: lorenzejay <lorenzejaytech@gmail.com>
105 lines
2.7 KiB
Python
105 lines
2.7 KiB
Python
from textwrap import dedent
|
|
|
|
import pytest
|
|
from mcp import StdioServerParameters
|
|
|
|
from crewai_tools import MCPServerAdapter
|
|
|
|
|
|
@pytest.fixture
|
|
def echo_server_script():
|
|
return dedent(
|
|
'''
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
mcp = FastMCP("Echo Server")
|
|
|
|
@mcp.tool()
|
|
def echo_tool(text: str) -> str:
|
|
"""Echo the input text"""
|
|
return f"Echo: {text}"
|
|
|
|
mcp.run()
|
|
'''
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def echo_server_sse_script():
|
|
return dedent(
|
|
'''
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
mcp = FastMCP("Echo Server", host="127.0.0.1", port=8000)
|
|
|
|
@mcp.tool()
|
|
def echo_tool(text: str) -> str:
|
|
"""Echo the input text"""
|
|
return f"Echo: {text}"
|
|
|
|
mcp.run("sse")
|
|
'''
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def echo_sse_server(echo_server_sse_script):
|
|
import subprocess
|
|
import time
|
|
|
|
# Start the SSE server process with its own process group
|
|
process = subprocess.Popen(
|
|
["python", "-c", echo_server_sse_script],
|
|
)
|
|
|
|
# Give the server a moment to start up
|
|
time.sleep(1)
|
|
|
|
try:
|
|
yield {"url": "http://127.0.0.1:8000/sse"}
|
|
finally:
|
|
# Clean up the process when test is done
|
|
process.kill()
|
|
process.wait()
|
|
|
|
|
|
def test_context_manager_syntax(echo_server_script):
|
|
serverparams = StdioServerParameters(
|
|
command="uv", args=["run", "python", "-c", echo_server_script]
|
|
)
|
|
with MCPServerAdapter(serverparams) as tools:
|
|
assert len(tools) == 1
|
|
assert tools[0].name == "echo_tool"
|
|
assert tools[0].run(text="hello") == "Echo: hello"
|
|
|
|
def test_context_manager_syntax_sse(echo_sse_server):
|
|
sse_serverparams = echo_sse_server
|
|
with MCPServerAdapter(sse_serverparams) as tools:
|
|
assert len(tools) == 1
|
|
assert tools[0].name == "echo_tool"
|
|
assert tools[0].run(text="hello") == "Echo: hello"
|
|
|
|
def test_try_finally_syntax(echo_server_script):
|
|
serverparams = StdioServerParameters(
|
|
command="uv", args=["run", "python", "-c", echo_server_script]
|
|
)
|
|
try:
|
|
mcp_server_adapter = MCPServerAdapter(serverparams)
|
|
tools = mcp_server_adapter.tools
|
|
assert len(tools) == 1
|
|
assert tools[0].name == "echo_tool"
|
|
assert tools[0].run(text="hello") == "Echo: hello"
|
|
finally:
|
|
mcp_server_adapter.stop()
|
|
|
|
def test_try_finally_syntax_sse(echo_sse_server):
|
|
sse_serverparams = echo_sse_server
|
|
mcp_server_adapter = MCPServerAdapter(sse_serverparams)
|
|
try:
|
|
tools = mcp_server_adapter.tools
|
|
assert len(tools) == 1
|
|
assert tools[0].name == "echo_tool"
|
|
assert tools[0].run(text="hello") == "Echo: hello"
|
|
finally:
|
|
mcp_server_adapter.stop()
|