Google UCP e Anthropic MCP: Integrando Protocolos
Salve galera, vim aqui compartilhar um projeto que eu fiz, integrando o UCP da google e o MCP da Anthropic,pedi pra IA dar uma ajuda no texto, mas é para fins de melhor entendimento, espero que usem e testem o repo e me deem algum feedback.
A Transição para a Economia de Agentes
A Inteligência Artificial Generativa está evoluindo de uma interface de consulta textual para um ecossistema de execução autônoma. No entanto, a viabilização do Comércio Agêntico (Agentic Commerce) esbarra em uma infraestrutura digital fragmentada. Historicamente, a integração entre plataformas de tecnologia e varejistas exigia um esforço manual e exaustivo de codificação. O UCP Nexus surge para colapsar essa complexidade, unindo o Universal Commerce Protocol (UCP) do Google e o Model Context Protocol (MCP) da Anthropic em uma arquitetura de middleware que transforma a web comercial em um ambiente nativo para IAs.
O Gargalo de Integração "N x N"
Antes do surgimento de padrões abertos, o comércio eletrônico operava sob o modelo de integrações ponto a ponto. No cenário de agentes de IA, isso criava o chamado gargalo "N x N": se houvesse 100 agentes diferentes e 100 lojas varejistas, seriam necessárias 10.000 integrações personalizadas para que todos interoperassem.
Cada loja possui sua própria lógica de carrinho, esquemas de checkout e APIs de frete. Para um agente de IA, navegar nessa heterogeneidade exigia um "exército de adaptadores" de código, tornando a escalabilidade financeiramente inviável e tecnicamente frágil. Sem uma linguagem comum, o custo de tokens para "ensinar" a estrutura de cada API ao modelo em tempo de execução saturaria qualquer janela de contexto.
Google UCP
O Universal Commerce Protocol (UCP) é um padrão aberto projetado para ser a linguagem comum. Ele resolve a fragmentação ao atuar como uma camada de abstração que padroniza como as capacidades comerciais são expostas e consumidas.
Arquitetura e Descoberta Dinâmica
A grande inovação do UCP reside na sua Descoberta Dinâmica. As empresas publicam um perfil padronizado em um endpoint específico: /.well-known/ucp
Zero Configuração: Quando um agente acessa uma loja, ele lê este arquivo JSON, identifica as capacidades suportadas (como checkout, descontos ou rastreamento) e se configura autonomamente.
Modularidade: O protocolo separa o comércio em Capacidades (núcleo, como Checkout) e Extensões (módulos como Fidelidade ou Fulfillment), permitindo uma implementação flexível e leve.
No UCP Nexus, a descoberta dinâmica é implementada através do UCPClient, que realiza a validação estrita contra os modelos oficiais do SDK:
class UCPClient:
def __init__(self, timeout: float = None, agent_profile: str = "default-hub-profile"):
if timeout is None:
timeout = settings.http_timeout
self.headers = {"UCP-Agent": f"profile={agent_profile}"}
self.client = httpx.Client(timeout=timeout, headers=self.headers)
def discover_services(self, url: str) -> UcpDiscoveryProfile:
base_url = url.rstrip("/")
discovery_url = f"{base_url}/.well-known/ucp"
try:
response = self.client.get(discovery_url)
response.raise_for_status()
except httpx.HTTPError as e:
raise UCPDiscoveryError(f"Failed to fetch discovery info from {discovery_url}: {e}") from e
try:
# Validate against the official SDK model
return UcpDiscoveryProfile(**response.json())
except ValidationError as e:
raise UCPConformanceError(f"Server response violated UCP Schema: {e}") from e
Utilizando uma analogia de infraestrutura, o UCP funciona como a padronização global de tomadas e voltagem. Antes dele, cada varejista tinha um "formato de tomada" proprietário; com o UCP, qualquer "robô" (agente de IA) pode se conectar instantaneamente a qualquer "loja" (negócio) e realizar a transação de forma segura, sem a necessidade de adaptadores customizados.
UCP Nexus como Servidor Proxy Inteligente
O projeto UCP Nexus funciona como um Servidor MCP Proxy que mapeia as capacidades UCP 1:1 para ferramentas MCP. Ele não processa pagamentos ou estoque, mas gerencia a comunicação e a integridade dos dados entre o LLM e o servidor do lojista.
Tradução e Mapeamento de Ferramentas
O Hub consome a especificação OpenRPC do lojista e a expõe como definições de ferramentas MCP. Para garantir a robustez, o Hub utiliza o UCP Python SDK, empregando modelos Pydantic gerados automaticamente para validar se o LLM enviou parâmetros corretos (como line_items e buyer_info) antes de encaminhar a requisição.
O registro de capacidades é gerenciado pelo ToolRegistry, que implementa o padrão de Deferred Loading:
class ToolRegistry:
def __init__(self):
self._deferred_tools: Dict[str, Discovery] = {}
self._loaded_tools: Dict[str, Discovery] = {}
def register_from_profile(self, profile: UcpDiscoveryProfile):
if profile.ucp and profile.ucp.capabilities:
for cap in profile.ucp.capabilities:
# Store by name (e.g., 'dev.ucp.shopping.checkout')
self._deferred_tools[cap.name] = cap
Eficiência de Contexto e Progressive Disclosure
Um dos maiores desafios da IA transacional é o consumo de tokens. Expor todas as ferramentas de uma loja simultaneamente pode consumir até 134 mil tokens antes da primeira interação. O Hub resolve isso habilitando o padrão de Progressive Disclosure (Divulgação Progressiva).
Deferred Loading: O Hub registra as ferramentas do lojista com uma flag de carregamento diferido.
Habilitação de Busca On-demand: O Hub expõe inicialmente apenas uma ferramenta de busca (tool_search). A IA, então, decide quando pesquisar por capacidades específicas (ex: "checkout"). Somente neste momento o Hub expande o schema pesado no contexto do modelo, preservando até 95% da janela de contexto original.
A implementação da ferramenta de busca é extremamente concisa:
class ToolSearchTool:
TOOL_NAME = "tool_search"
def __init__(self, registry: ToolRegistry):
self.registry = registry
@property
def definition(self) -> Tool:
return Tool(
name=self.TOOL_NAME,
description="Search for tools using regex.",
inputSchema={
"type": "object",
"properties": {
"regex": {"type": "string"}
},
"required": ["regex"]
}
)
async def execute(self, arguments: dict) -> list[Any]:
regex = arguments.get("regex")
if not regex: return []
return self.registry.search_tools(regex)
O servidor MCP expõe essa capacidade de forma declarativa através do FastMCP:
mcp = FastMCP("UCP-to-MCP Hub")
registry = ToolRegistry()
search_tool = ToolSearchTool(registry)
code_tool = CodeExecutionTool(registry)
@mcp.tool(name="tool_search")
async def search_tools(regex: str) -> list[dict]:
return await search_tool.execute({"regex": regex})
@mcp.tool()
async def refresh_ucp_discovery(url: str) -> str:
client = UCPClient()
try:
profile = client.discover_services(url)
registry.register_from_profile(profile)
count = len(registry._deferred_tools)
return f"Successfully discovered {count} capabilities from {url}."
except Exception as e:
return f"Discovery failed: {e}"
Orquestração via Code Mode (Sandbox de Execução)
Uma das inovações críticas do projeto é a implementação da capacidade de Code Execution do MCP. Em vez de forçar o Agente a realizar múltiplas chamadas individuais de ferramentas (o que elevaria a latência e o custo), o Hub habilita o Agente a orquestrar fluxos complexos escrevendo scripts Python.
Performance e Redução de Latência
Ao prover um ambiente seguro (Sandbox), o Hub permite que o Agente escreva código para realizar tarefas massivas localmente, como filtrar milhares de itens de um catálogo e devolver apenas o sumário necessário.
Execução Paralela: O Agente pode utilizar asyncio.gather para disparar simultaneamente chamadas de checkout e cálculo de impostos.
Papel do Hub vs. Agente: O Hub fornece o "Playground" e as ferramentas; o Agente (como o Claude) é quem decide a lógica, escreve o script e comanda a execução, reduzindo o round-trip entre a IA e a API.
A implementação do Sandbox fornece um ambiente restrito mas poderoso para execução:
class Sandbox:
def __init__(self, registry: ToolRegistry, additional_globals: dict = None):
self.proxy = UCPProxy(registry)
self.safe_globals = self._build_safe_globals(additional_globals)
def _build_safe_globals(self, additional_globals: dict = None) -> dict:
import importlib
base_globals = {
"ucp": self.proxy, "print": print,
"__builtins__": {
"print": print, "len": len, "range": range, "list": list,
"dict": dict, "set": set, "str": str, "int": int,
"float": float, "bool": bool, "next": next,
}
}
for module_name in settings.sandbox_globals:
try:
mod = importlib.import_module(module_name)
base_globals[module_name] = mod
except ImportError:
print(f"Warning: Could not import '{module_name}'")
if additional_globals:
safe = additional_globals.copy()
if "__builtins__" in safe: del safe["__builtins__"]
base_globals.update(safe)
return base_globals
async def run(self, code: str) -> str:
from io import StringIO
import contextlib
output_buffer = StringIO()
indented_code = "\n".join(" " + line for line in code.splitlines())
wrapped_code = f"async def _agent_script():\n{indented_code}"
try:
with contextlib.redirect_stdout(output_buffer):
exec(wrapped_code, self.safe_globals)
_agent_script = self.safe_globals["_agent_script"]
await _agent_script()
return output_buffer.getvalue()
except Exception as e:
return f"{output_buffer.getvalue()}\nRuntime Error: {e}"
O UCPProxy injetado no sandbox permite ao agente interagir com serviços UCP de forma segura:
class UCPProxy:
def __init__(self, registry: ToolRegistry):
self._client = UCPClient()
self._registry = registry
self._security = AP2Security()
self._discovered_payment_handlers = []
self._last_discovery_url = settings.ucp_server_url
async def discover(self, url: str) -> list[dict]:
profile = self._client.discover_services(url)
self._last_discovery_url = url.rstrip("/")
self._registry.register_from_profile(profile)
if profile.payment and profile.payment.handlers:
self._discovered_payment_handlers = profile.payment.handlers
results = []
if profile.ucp and profile.ucp.capabilities:
for cap in profile.ucp.capabilities:
results.append({
"name": cap.name,
"spec": str(cap.spec) if cap.spec else None,
"version": str(cap.version)
})
return results
async def call(self, tool_name: str, **kwargs) -> Any:
print(f"[UCPProxy] Calling tool: {tool_name}")
if not self._last_discovery_url:
raise RuntimeError("Must call 'await ucp.discover(url)' first.")
endpoint_path = self._resolve_endpoint(tool_name)
if not endpoint_path:
raise ValueError(f"Tool '{tool_name}' not mapped.")
full_url = f"{self._last_discovery_url}{endpoint_path}"
return self._dispatch_request(full_url, kwargs)
Negociação de Segurança e Pagamentos Autônomos
O Hub gerencia a "conversa técnica" de segurança entre o Agente e o lojista, garantindo que dados sensíveis nunca sejam expostos ao processamento de linguagem natural do LLM.
Arquitetura Desacoplada: O processo é dividido em Negociação (recebimento de handlers como Google Pay), Aquisição (obtenção de token do provedor) e Finalização (envio do token opaco ao lojista).
AP2 Mandates: Para operações 100% autônomas, o Hub suporta a extensão dev.ucp.shopping.ap2_mandate. Ele permite que o Agente envie mandatos criptográficos assinados com Ed25519, provando a autorização do usuário final para aquela transação específica, sem intervenção humana em tempo real.
A implementação do gerenciamento de chaves criptográficas usando Ed25519 (RFC 8032):
class KeyManager:
def __init__(self):
self._private_key = ed25519.Ed25519PrivateKey.generate()
self._public_key = self._private_key.public_key()
public_bytes = self._public_key.public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw
)
self.key_id = f"hub-key-{base64.urlsafe_b64encode(public_bytes[:8]).decode().rstrip('=')}"
def sign(self, payload: str) -> str:
signature = self._private_key.sign(payload.encode("utf-8"))
return base64.urlsafe_b64encode(signature).decode().rstrip("=")
def get_public_jwk(self) -> Dict[str, Any]:
public_bytes = self._public_key.public_bytes(
encoding=serialization.Encoding.Raw,
format=serialization.PublicFormat.Raw
)
x_coord = base64.urlsafe_b64encode(public_bytes).decode().rstrip("=")
return {"kty": "OKP", "crv": "Ed25519", "x": x_coord, "kid": self.key_id}
A criação de mandatos AP2 para pagamentos autônomos:
class AP2Security:
def __init__(self):
self.key_manager = KeyManager()
def create_mandate(self, amount: float, currency: str, beneficiary: str) -> str:
header = {"alg": "EdDSA", "typ": "JWT", "kid": self.key_manager.key_id}
payload = {
"iss": "ucp-hub-mcp",
"sub": "agent-autonomous-action",
"aud": beneficiary,
"exp": int(time.time()) + settings.jwt_expiry_seconds,
"scope": "ucp:payment",
"mandate": {"max_amount": amount, "currency": currency}
}
b64_header = base64.urlsafe_b64encode(json.dumps(header).encode()).decode().rstrip("=")
b64_payload = base64.urlsafe_b64encode(json.dumps(payload).encode()).decode().rstrip("=")
signing_input = f"{b64_header}.{b64_payload}"
signature = self.key_manager.sign(signing_input)
return f"{signing_input}.{signature}"
O proxy expõe a funcionalidade de seleção de método de pagamento para o agente:
async def select_payment_method(self, method_name: str, amount: float = 0.0, currency: str = "BRL") -> dict:
import uuid
mandate_jwt = self._security.create_mandate(amount, currency, beneficiary="merchant-id")
print(f"[UCPProxy] Generated AP2 Mandate for {amount} {currency} via {method_name}")
return {
"token": f"pay_{uuid.uuid4().hex[:24]}",
"mandate": mandate_jwt,
"method": method_name
}
O mecanismo de conformidade garante a integridade e autenticidade de cada requisição:
def _get_conformance_headers(self, payload_str: str = "") -> dict:
import uuid
import time
request_id = str(uuid.uuid4())
idempotency_key = str(uuid.uuid4())
timestamp = str(int(time.time()))
nonce = uuid.uuid4().hex[:16]
signing_input = f"{timestamp}.{nonce}.{payload_str}"
signature = self._key_manager.sign(signing_input)
return {
"request-id": request_id,
"idempotency-key": idempotency_key,
"ucp-timestamp": timestamp,
"ucp-nonce": nonce,
"request-signature": signature,
"ucp-key-id": self._key_manager.key_id
}
E finalmente, o dispatch de requisições HTTP com serialização canônica para garantir a validação de assinatura:
def _dispatch_request(self, url: str, kwargs: dict) -> dict:
import json
checkout_id = kwargs.get("id")
action = kwargs.pop("_action", None)
client = self._client.client
payload = kwargs
if action == "complete": payload = kwargs.get("payment", kwargs)
payload_str = json.dumps(payload, sort_keys=True)
headers = self._get_conformance_headers(payload_str)
try:
if not checkout_id:
# CREATE
print(f"[UCPProxy] Transport: POST {url}")
resp = client.post(url, content=payload_str, headers=headers)
elif action == "complete":
# COMPLETE
complete_url = f"{url}/{checkout_id}/complete"
print(f"[UCPProxy] Transport: POST {complete_url}")
resp = client.post(complete_url, content=payload_str, headers=headers)
else:
# UPDATE
update_url = f"{url}/{checkout_id}"
print(f"[UCPProxy] Transport: PUT {update_url}")
resp = client.put(update_url, content=payload_str, headers=headers)
resp.raise_for_status()
return resp.json()
except Exception as e:
self._handle_http_error(e)
Fases de Implementação e Validação de Conformidade
O desenvolvimento do Hub seguiu um plano de ação rigoroso, focado na fidelidade aos padrões oficiais:
-
Discovery Dinâmico: Implementação do mecanismo de GET em
/.well-known/ucppara identificação de serviços (REST ou MCP). -
Scaffolding de Modelos: Uso estrito do SDK oficial para garantir que todos os campos de data sigam o RFC 3339 e valores monetários sejam processados em unidades menores (centavos).
-
Teste de Conformidade (Conformance): O Hub foi validado contra o servidor de referência
flower_shop. Utilizou-se a suíte de testes de integração (checkout_lifecycle_test.pyeidempotency_test.py) para verificar se o Hub mantém a integridade do estado da transação e lida corretamente com falhas de rede.
Conclusão
O UCP Nexus habilita capacidades avançadas como execução de código em sandbox e carregamento diferido de schemas. Resultando em um ecosistema para a IA.