ACP-MCP-Server

Instructions

Implementation Reference

• acp_mcp_server/message_bridge.py:134-167 (handler)

  The @mcp.tool()-decorated async handler function that implements the core logic of the 'analyze_message_content' tool. It parses the input acp_message_json into an ACPMessage, analyzes the parts (total count, content types, presence of URLs, encodings, total content size), and returns a structured JSON analysis.

  @mcp.tool()
  async def analyze_message_content(acp_message_json: str) -> str:
      """Analyze the content types and structure of an ACP message"""
      try:
          import json
          message_data = json.loads(acp_message_json)
          acp_message = ACPMessage(**message_data)
          
          analysis = {
              "total_parts": len(acp_message.parts),
              "content_types": {},
              "has_urls": False,
              "encodings": set(),
              "total_size": 0
          }
          
          for part in acp_message.parts:
              content_type = part.content_type
              analysis["content_types"][content_type] = analysis["content_types"].get(content_type, 0) + 1
              analysis["encodings"].add(part.content_encoding)
              
              if part.content_url:
                  analysis["has_urls"] = True
              
              if part.content:
                  analysis["total_size"] += len(part.content)
          
          analysis["encodings"] = list(analysis["encodings"])
          
          return json.dumps(analysis, indent=2)
          
      except Exception as e:
          return f"Error: {e}"

• acp_mcp_server/message_bridge.py:8-17 (schema)

  Pydantic BaseModel definitions for ACPMessagePart (lines 8-14) and ACPMessage (lines 15-17), which provide input validation and type parsing for the ACP message structure used in the analyze_message_content handler.

  class ACPMessagePart(BaseModel):
      name: Optional[str] = None
      content_type: str
      content: Optional[str] = None
      content_encoding: Optional[str] = "plain"
      content_url: Optional[str] = None
  
  class ACPMessage(BaseModel):
      parts: List[ACPMessagePart]

• acp_mcp_server/server.py:86-86 (registration)

  Invocation of register_bridge_tools(self.mcp, self.message_bridge) during server initialization, which executes the tool registrations including the @mcp.tool() decorator for analyze_message_content.

  register_bridge_tools(self.mcp, self.message_bridge)

• acp_mcp_server/message_bridge.py:117-117 (registration)

  The register_bridge_tools function definition that contains the @mcp.tool() registrations for bridge tools, including analyze_message_content and convert_acp_message.

  def register_bridge_tools(mcp: FastMCP, bridge: MessageBridge):