Authorization: Bearer <your-jwt-token>

{
  "jsonrpc": "2.0",
  "method": "string",
  "params": {},
  "id": "string"
}

# Get authentication token
TOKEN=$(curl -s -X POST "https://mcpairplatformapim01.azure-api.net/agent-gallery-api-gateway/login" \
    -H "Content-Type: application/x-www-form-urlencoded" \
    -d "username=qi.chang@accenture.com&password=test12345" \
  | jq -r .access_token)

# Launch Math-MCP agent
DEPLOYMENT_ID=$(curl -s -X POST "https://mcpairplatformapim01.azure-api.net/agent-gallery-api-gateway/launch" \
    -H "Authorization: Bearer $TOKEN" \
    -H "Content-Type: application/json" \
    -d '{
      "srcOwner": "EthanHenrickson",
      "srcRepo": "math-mcp",
      "srcAgentType": "Third-Party"
    }' | jq -r .data.deploymentId)

import asyncio
import json
from typing import Any, Dict, Optional
import httpx

class AgentGalleryMCPClient:
    """FastMCP-style SSE client for Agent Gallery API with full response handling"""

    def __init__(self, base_url: str, deployment_id: str, token: str):
        self.base_url = base_url
        self.deployment_id = deployment_id
        self.token = token
        self.session_id: Optional[str] = None
        self.sse_url = f"{base_url}/{deployment_id}/sse"
        self.response_queue = asyncio.Queue()
        self.listener_task = None

    async def connect(self):
        """Connect and start listening to SSE with MCP initialization"""
        print("Connecting to SSE endpoint...")
        
        self.listener_task = asyncio.create_task(self._sse_listener())
        
        for _ in range(50):
            if self.session_id:
                print(f"Connected with session_id: {self.session_id}")
                break
            await asyncio.sleep(0.1)
        
        if not self.session_id:
            raise Exception("Failed to get session_id")
        
        print("Initializing MCP protocol...")
        init_response = await self.send_and_wait(
            "initialize",
            {
                "protocolVersion": "0.1.0",
                "capabilities": {}
            },
            "init-1",
            timeout=10
        )
        
        if "result" in init_response:
            server_info = init_response['result'].get('serverInfo', {})
            name = server_info.get('name', 'unknown')
            version = server_info.get('version', 'unknown')
            print(f"Initialized: {name} v{version}")
        else:
            print("WARNING: Initialization response unexpected:", init_response)

    async def _sse_listener(self):
        """Background SSE listener that handles event/data format"""
        async with httpx.AsyncClient(timeout=None) as client:
            async with client.stream(
                "GET",
                self.sse_url,
                headers={
                    "Accept": "text/event-stream",
                    "Authorization": f"Bearer {self.token}"
                }
            ) as response:
                current_event = None
                async for line in response.aiter_lines():
                    if line.startswith(": heartbeat"):
                        continue
                        
                    if line and line != ": heartbeat":
                        if not self.session_id and "session_id=" in line:
                            self.session_id = line.split("session_id=")[-1].strip()
                        
                        if line.startswith("event:"):
                            current_event = line[6:].strip()
                        
                        elif line.startswith("data: ") and line != "data: null":
                            data_str = line[6:]
                            
                            if current_event == "message":
                                try:
                                    data = json.loads(data_str)
                                    await self.response_queue.put(data)
                                except json.JSONDecodeError:
                                    pass
                            
                            current_event = None

    async def send_and_wait(
        self, method: str, params: Dict[str, Any] = None, 
        request_id: str = None, timeout: float = 15
    ) -> Any:
        """Send request and wait for response with matching ID"""
        if not self.session_id:
            raise Exception("Not connected")

        request_id = request_id or f"{method}-{asyncio.get_event_loop().time()}"
        request = {
            "jsonrpc": "2.0",
            "method": method,
            "params": params or {},
            "id": request_id
        }

        print(f"\nSending {method} (id={request_id})...")
        
        async with httpx.AsyncClient(timeout=30.0) as client:
            response = await client.post(
                f"{self.sse_url}?session_id={self.session_id}",
                headers={
                    "Authorization": f"Bearer {self.token}",
                    "Content-Type": "application/json",
                    "Accept": "text/event-stream"
                },
                json=request
            )
            
            if response.status_code != 200:
                raise Exception(f"Request failed: {response.status_code}")

        start_time = asyncio.get_event_loop().time()
        checked_messages = []
        
        while asyncio.get_event_loop().time() - start_time < timeout:
            try:
                data = await asyncio.wait_for(self.response_queue.get(), timeout=0.5)
                
                if data.get("id") == request_id:
                    return data
                else:
                    checked_messages.append(data)
                    
            except asyncio.TimeoutError:
                continue
        
        for msg in checked_messages:
            await self.response_queue.put(msg)
        
        raise TimeoutError(f"No response received for {request_id} after {timeout}s")

    async def list_tools(self) -> list:
        """List tools and return the actual list - FastMCP style"""
        response = await self.send_and_wait("tools/list", {}, "tools-list")
        
        if "result" in response and "tools" in response["result"]:
            tools = response["result"]["tools"]
            print(f"Received {len(tools)} tools")
            return tools
        else:
            print("ERROR: No tools in response")
            return []

    async def call_tool(self, name: str, arguments: Dict[str, Any]) -> str:
        """Call a tool and return the result - FastMCP style"""
        params = {
            "name": name,
            "arguments": arguments
        }
        
        response = await self.send_and_wait("tools/call", params, f"{name}-call")
        
        if "result" in response and "content" in response["result"]:
            for content in response["result"]["content"]:
                if content.get("type") == "text":
                    result = content["text"]
                    print(f"Result: {result}")
                    return result
        
        if "error" in response:
            error = response["error"]
            print(f"ERROR: {error}")
            raise Exception(f"Tool call failed: {error}")
        
        print("ERROR: No result in response")
        return None

    async def close(self):
        """Clean up resources"""
        if self.listener_task:
            self.listener_task.cancel()
            try:
                await self.listener_task
            except asyncio.CancelledError:
                pass
        print("\nConnection closed")

client = AgentGalleryMCPClient(<HOST>/sse, <DEPLOYMENT_ID>, <TOKEN>)

# Connect MCP server
await client.connect()

# List tools
tools = await client.list_tools()

# Call add tool
result = await client.call_tool("add", {
    "firstNumber": 10,
    "secondNumber": 20
})