MAGG

Instructions

Implementation Reference

• magg/proxy/mixin.py:66-172 (handler)

  The core handler function for the 'proxy' tool. It validates inputs and dispatches to _proxy_list, _proxy_info, or _proxy_call based on the action, using the backend client to interact with proxied MCP capabilities.

  async def _proxy_tool(
      self,
      action: Annotated[LiteralProxyAction, Field(
          description="Action to perform: list, info, or call."
      )],
      a_type: Annotated[LiteralProxyType, Field(
          description="Type of MCP capability to interact with: tool, resource, or prompt.",
          alias="type"
      )],
      args: Annotated[dict[str, Any] | str | None, Field(
          description="Arguments for a 'call' action (call tool, read resource, or get prompt). Can be a dict or JSON string."
      )] = None,
      path: Annotated[str | None, Field(
          description="Name or URI of the specific tool/resource/prompt (with FastMCP prefixing).\n"
                      "Not allowed for 'list' and 'info' actions.",
          # validation_alias=AliasChoices("name", "uri"),
      )] = None,
      limit: Annotated[int | None, Field(
          description="Maximum number of items to return (for 'list' action only). Default: 100",
          ge=1,
          le=1000
      )] = None,
      offset: Annotated[int | None, Field(
          description="Number of items to skip (for 'list' action only). Default: 0",
          ge=0
      )] = None,
      filter_server: Annotated[str | None, Field(
          description="Filter results by server name prefix (for 'list' action only)"
      )] = None,
  ) -> Any:
      """Main proxy tool for dynamic access to mounted MCP servers.
  
      This tool provides a unified interface for:
      - Listing available tools, resources, or prompts across servers
      - Getting detailed info about specific capabilities
      - Calling tools, reading resources, or getting prompts
  
      Annotations are used to provide rich type information for results,
      which can generally be expected to ultimately include JSON-encoded
      EmbeddedResource results that can be interpreted by the client.
      """
      self.validate_operation(action=action, a_type=a_type)
  
      if action in frozenset({"info", "call"}) and not path:
          raise ValueError(
              f"Parameter 'path' is required for action {action!r}"
          )
  
      if action in frozenset({"list", "info"}) and args:
          raise ValueError(
              f"Parameter 'args' should not be provided for action {action!r}"
          )
  
      if action == "list" and path:
          raise ValueError(
              "Parameter 'path' should not be provided for action 'list'"
          )
  
      if action != "list" and (limit is not None or offset is not None or filter_server is not None):
          raise ValueError(
              "Parameters 'limit', 'offset', and 'filter_server' are only allowed for 'list' action"
          )
  
      if action == "list":
          result, result_type = await self._proxy_list(a_type)
  
          if filter_server and result:
              result = [item for item in result if hasattr(item, 'name') and item.name.startswith(filter_server)]
  
          total_count = len(result) if result else 0
          if result and (limit or offset):
              offset_val = offset or 0
              limit_val = limit or 100
              result = result[offset_val:offset_val + limit_val]
  
          if result:
              result = embed_python_object_list_in_resource(
                  typ=result_type,
                  obj=result,
                  uri=f"{self.PROXY_TOOL_NAME}:{action}/{a_type}",
                  proxyAction=action,
                  proxyType=a_type,
                  totalCount=total_count,
                  offset=offset or 0,
                  limit=limit or 100,
              )
  
      elif action == "info":
          result = await self._proxy_info(a_type, path)
  
          if result:
              result = embed_python_object_in_resource(
                  obj=result,
                  uri=f"{self.PROXY_TOOL_NAME}:{action}/{a_type}/{path}",
                  proxyAction=action,
                  proxyType=a_type,
                  proxyPath=path,
              )
  
      elif action == "call":
          result = await self._proxy_call(a_type, path, json_to_dict(args))
      else:
          raise ValueError(
              f"Unknown action: {action!r}. Supported actions are 'list', 'info', and 'call'."
          )
  
      return result

• magg/proxy/server.py:83-90 (registration)

  Registers the 'proxy' tool on the FastMCP server by creating a FunctionTool from the _proxy_tool handler with name PROXY_TOOL_NAME ("proxy") and adding it via add_tool.

  def _register_proxy_tool(self):
      tool = FunctionTool.from_function(
          self._proxy_tool,
          name=self.PROXY_TOOL_NAME,
          serializer=self._tool_serializer,
      )
  
      self.add_tool(tool)

• magg/proxy/types.py:8-9 (schema)

  Pydantic Literal type definitions for the 'type' and 'action' parameters in the proxy tool schema.

  LiteralProxyType = Literal["tool", "resource", "prompt"]
  LiteralProxyAction = Literal["list", "info", "call"]

• magg/proxy/mixin.py:36-36 (registration)

  Defines the tool name constant PROXY_TOOL_NAME = "proxy" used in registration.

  PROXY_TOOL_NAME: ClassVar[str] = "proxy"

• magg/server/manager.py:55-60 (registration)

  Instantiates the ProxyFastMCP server instance, which via ProxyMCP mixin __init__ calls _register_proxy_tool to register the proxy tool.

  self.mcp = ProxyFastMCP(
      name=self.self_prefix,
      instructions=MAGG_INSTRUCTIONS.format(self_prefix=self.self_prefix),
      auth=auth_provider
  )
  self.mounted_servers = {}