hello, could i add some pre/post say method to too...
# agents
f
hello, could i add some pre/post say method to tools using mcp? i know it was ok to added to native tools , not sure if there's support for mcp ones
l
Interesting, I don't think there is a simple interface for this. I got pre working with a custom MCP server, but post say would require tracking when the agent finishes speaking:
Copy code
class 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>")
f
yes i came up with something similar, pre say is ok now ( but i observe the voice is a bit late , looks like coming out when tool finish)
Copy code
def _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,
      },
    )