File size: 2,252 Bytes
b39a029 fad6056 19b980e 2f16687 7886f2b 2f16687 19b980e 44b61df 19b980e 44b61df 19b980e 38c4b14 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
# References:
# https://github.com/modelcontextprotocol/python-sdk
import anyio, asyncio, os
from mcp.client.sse import sse_client
from mcp.client.session import ClientSession, DEFAULT_CLIENT_INFO
# Chess position evaluation
SPACE_AUTHOR_NAME = os.getenv("SPACE_AUTHOR_NAME")
MCP_SSE_URL_CHESS_POSITION_EVALUATION = f"https://{SPACE_AUTHOR_NAME}-chess.hf.space/gradio_api/mcp/sse"
MCP_TOOL_CHESS_POSITION_EVALUATION = "chess_position_evaluation"
async def call_mcp_tool_async(mcp_url: str, tool_name: str, arguments: dict, timeout: int = 5) -> str:
"""Call an MCP tool asynchronously.
Args:
mcp_url (str): The MCP SSE URL
tool_name (str): The name of the MCP tool to call
arguments (dict): The arguments to pass to the tool
timeout (int): Timeout in seconds (default: 5)
Returns:
str: The tool result as a string
Raises:
Exception: If the MCP call fails
"""
try:
with anyio.fail_after(timeout):
async with sse_client(mcp_url) as (read, write):
async with ClientSession(read, write, client_info=DEFAULT_CLIENT_INFO) as session:
await session.initialize()
result = await session.call_tool(tool_name, arguments=arguments)
if result.content:
text_parts = [
item.text for item in result.content
if hasattr(item, 'text')
]
return " ".join(text_parts)
return str(result)
except Exception as e:
raise
def call_mcp_tool(mcp_url: str, tool_name: str, arguments: dict, timeout: int = 30) -> str:
"""Call an MCP tool synchronously.
Args:
mcp_url (str): The MCP SSE URL
tool_name (str): The name of the MCP tool to call
arguments (dict): The arguments to pass to the tool
timeout (int): Timeout in seconds (default: 30)
Returns:
str: The tool result as a string
Raises:
Exception: If the MCP call fails
"""
return asyncio.run(call_mcp_tool_async(mcp_url, tool_name, arguments, timeout)) |