few-analyst-46
12/08/2025, 1:07 AMlively-winter-88668
12/08/2025, 10:53 AMclass MyMCPServer(mcp.MCPServerHTTP):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Reference to the session (set after session is created)
self.session: AgentSession | None = None
def _make_function_tool(self, name, description, input_schema, meta):
# Get the original tool (this is the callable function itself with metadata attached)
original_tool = super()._make_function_tool(name, description, input_schema, meta)
# Create a wrapper that adds before/after hooks
async def wrapped_tool(raw_arguments: dict):
# Say something BEFORE the MCP call
if self.session:
self.session.say("Looking that up for you.")
try:
# Call the MCP tool
result = await original_tool(raw_arguments)
return result
except Exception as e:
logger.error(f"MCP call {name} failed: {e}")
raise
# Return a new tool with the wrapper
from livekit.agents import function_tool
return function_tool(wrapped_tool, raw_schema={
"name": name,
"description": description,
"parameters": input_schema,
**({"meta": meta} if meta else {})
})
mcp_server = MyMCPServer(url="<https://docs.livekit.io/mcp>")few-analyst-46
12/08/2025, 11:56 AMdef _make_function_tool(
self, name: str, description: str | None, input_schema
):
# Get callbacks for this specific tool
callbacks = self._tool_callbacks.get(name, {})
pre_callback = callbacks.get('pre', defualt_pre_callback)
post_callback = callbacks.get('post', default_post_callback)
async def _tool_called(raw_arguments: dict):
# In case (somehow), the tool is called after the MCPServer aclose.
if self._client is None:
raise mcp.ToolError(
'Tool invocation failed: internal service is unavailable. '
'Please check that the MCPServer is still running.'
)
# Execute pre-callback if defined - returns (arguments, meta) tuple
arguments = raw_arguments
call_meta = self._meta
try:
arguments, call_meta = await pre_callback(name, arguments, self._meta)
except Exception as e:
logger.exception('pre-callback error for tool %s: %s', name, str(e))
raise mcp.ToolError(f"Tool '{name}' pre failed") from e
# TODO(Deo): try fixing <https://github.com/livekit/agents/issues/2429>
tool_result = None
for _ in range(2):
try:
tool_result = await self._client.call_tool(
name, arguments, meta=call_meta
)
except anyio.ClosedResourceError:
logger.error('mcp client error, reinitializing')
await self.initialize()
await self.list_tools()
continue
break
if tool_result is None:
raise mcp.ToolError(
f"Tool '{name}' call failed after retries. "
'This might indicate a connection issue.'
)
if tool_result.isError:
error_str = '\n'.join(str(part) for part in tool_result.content)
raise mcp.ToolError(error_str)
try:
tool_result: CallToolResult | None = await post_callback(
name, arguments, call_meta, tool_result
)
except Exception as e:
logger.exception('post-callback error for tool %s: %s', name, str(e))
raise mcp.ToolError(f"Tool '{name}' post processing failed") from e
if tool_result is None:
# None means do nothing, should be handled by post-callback
return
# Serialize result after post-callback processing
if len(tool_result.content) == 1:
return tool_result.content[0].model_dump_json()
elif len(tool_result.content) > 1:
return orjson.dumps(
[item.model_dump() for item in tool_result.content]
).decode()
raise mcp.ToolError(
f"Tool '{name}' completed without producing a result. "
'This might indicate an issue with internal processing.'
)
return function_tool(
_tool_called,
raw_schema={
'name': name,
'description': description,
'parameters': input_schema,
},
)