MCP Protocol

# From mcp_server/src/pricing_mcp/mcp_server.py:108-149

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    settings.mcp_server_name,
    host=settings.http_host,
    port=settings.http_port,
)

@mcp.tool()
async def optimal(
    pricing_url: Optional[str] = None,
    pricing_yaml: Optional[str] = None,
    filters: Optional[Dict[str, Any]] = None,
    solver: str = "minizinc",
    objective: str = "minimize",
    refresh: bool = False,
) -> Dict[str, Any]:
    """Compute the optimal subscription under the provided constraints."""
    
    if not (pricing_url or pricing_yaml):
        raise ValueError(
            "optimal requires pricing_url or pricing_yaml to run analysis."
        )
    
    if solver not in VALID_SOLVERS:
        raise ValueError(INVALID_SOLVER_ERROR)
    
    if objective not in {"minimize", "maximize"}:
        raise ValueError("objective must be 'minimize' or 'maximize'.")
    
    logger.info(
        TOOL_INVOKED,
        tool="optimal",
        pricing_url=pricing_url,
        has_pricing_yaml=bool(pricing_yaml),
        filters=filters,
        solver=solver,
        objective=objective,
        refresh=refresh,
    )
    
    result = await container.workflow.run_optimal(
        url=pricing_url or "",
        filters=filters,
        solver=solver,
        objective=objective,
        refresh=refresh,
        yaml_content=pricing_yaml,
    )
    
    logger.info(TOOL_COMPLETED, tool="optimal", keys=list(result.keys()))
    return result

# From mcp_server/src/pricing_mcp/workflows/pricing.py (conceptual)

class PricingWorkflow:
    def __init__(self, amint_client, analysis_client, cache):
        self.amint = amint_client
        self.analysis = analysis_client
        self.cache = cache
    
    async def run_optimal(
        self,
        url: Optional[str],
        yaml_content: Optional[str],
        filters: Optional[Dict],
        solver: str,
        objective: str,
        refresh: bool,
    ) -> Dict:
        # 1. Get Pricing2Yaml content
        if yaml_content:
            pricing_yaml = yaml_content
        elif url:
            cache_key = f"pricing:{url}"
            if not refresh:
                cached = await self.cache.get(cache_key)
                if cached:
                    pricing_yaml = cached
                else:
                    pricing_yaml = await self.amint.transform(url)
                    await self.cache.set(cache_key, pricing_yaml)
            else:
                pricing_yaml = await self.amint.transform(url)
                await self.cache.set(cache_key, pricing_yaml)
        else:
            raise ValueError("Must provide url or yaml_content")
        
        # 2. Call Analysis API
        result = await self.analysis.optimal(
            pricing_yaml=pricing_yaml,
            filters=filters,
            solver=solver,
            objective=objective,
        )
        
        # 3. Return structured result
        return result

# User calls optimal without url or yaml
raise ValueError(
    "optimal requires pricing_url or pricing_yaml to run analysis."
)

# MCP server responds:
{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {
    "code": -32602,  # Invalid params
    "message": "optimal requires pricing_url or pricing_yaml to run analysis."
  }
}

{
  "timestamp": "2024-07-02T10:30:00Z",
  "level": "INFO",
  "event": "mcp.tool.invoked",
  "tool": "optimal",
  "pricing_url": "https://buffer.com/pricing",
  "filters": {"features": ["hashtagManager"]},
  "solver": "minizinc",
  "objective": "minimize"
}

pip install mcp