431 lines
19 KiB
Python
431 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from deepagents import create_deep_agent
|
|
from langchain_core.messages import BaseMessage
|
|
from langchain_core.tools import StructuredTool
|
|
from langchain_openai import ChatOpenAI
|
|
from pydantic import BaseModel, Field
|
|
|
|
from a2a_pack import (
|
|
A2AAgent,
|
|
ConsumerSetup,
|
|
ConsumerSetupField,
|
|
ConsumerSetupMissing,
|
|
EgressPolicy,
|
|
LLMProvisioning,
|
|
Pricing,
|
|
Resources,
|
|
RunContext,
|
|
skill,
|
|
)
|
|
|
|
|
|
DEFAULT_BASE_URL = "https://blog.a2acloud.io"
|
|
OPERATIONS = json.loads("{\n \"create_blog_post\": {\n \"description\": \"Create a blog post\",\n \"destructive\": true,\n \"method\": \"POST\",\n \"operation_id\": \"create_blog_post\",\n \"parameters\": [],\n \"path\": \"/api/posts\",\n \"request_body\": {\n \"$ref\": \"#/components/schemas/BlogPostInput\"\n },\n \"security\": [\n {\n \"bearerAuth\": []\n },\n {\n \"apiKeyAuth\": []\n }\n ],\n \"skill_name\": \"create_blog_post\",\n \"summary\": \"Create a blog post\",\n \"tags\": [\n \"Posts\"\n ]\n },\n \"delete_blog_post\": {\n \"description\": \"Delete a post by slug\",\n \"destructive\": true,\n \"method\": \"DELETE\",\n \"operation_id\": \"delete_blog_post\",\n \"parameters\": [],\n \"path\": \"/api/posts/{slug}\",\n \"request_body\": null,\n \"security\": [\n {\n \"bearerAuth\": []\n },\n {\n \"apiKeyAuth\": []\n }\n ],\n \"skill_name\": \"delete_blog_post\",\n \"summary\": \"Delete a post by slug\",\n \"tags\": [\n \"Posts\"\n ]\n },\n \"get_blog_health\": {\n \"description\": \"Check blog service health\",\n \"destructive\": false,\n \"method\": \"GET\",\n \"operation_id\": \"get_blog_health\",\n \"parameters\": [],\n \"path\": \"/api/healthz\",\n \"request_body\": null,\n \"security\": null,\n \"skill_name\": \"get_blog_health\",\n \"summary\": \"Check blog service health\",\n \"tags\": [\n \"Health\"\n ]\n },\n \"get_blog_post\": {\n \"description\": \"Returns a published post. Add `preview=1` to fetch drafts or future-dated posts; preview requests require either bearer auth or the `x-api-key` header.\",\n \"destructive\": false,\n \"method\": \"GET\",\n \"operation_id\": \"get_blog_post\",\n \"parameters\": [\n {\n \"description\": \"Set to `1` to bypass the public published-post filter. Requires authentication.\",\n \"in\": \"query\",\n \"name\": \"preview\",\n \"required\": false,\n \"schema\": {\n \"enum\": [\n \"1\"\n ],\n \"type\": \"string\"\n }\n }\n ],\n \"path\": \"/api/posts/{slug}\",\n \"request_body\": null,\n \"security\": [\n {},\n {\n \"bearerAuth\": []\n },\n {\n \"apiKeyAuth\": []\n }\n ],\n \"skill_name\": \"get_blog_post\",\n \"summary\": \"Get a post by slug\",\n \"tags\": [\n \"Posts\"\n ]\n },\n \"list_blog_posts\": {\n \"description\": \"Returns published posts by default. `status=all` and `status=draft` include draft content and require either bearer auth or the `x-api-key` header.\",\n \"destructive\": false,\n \"method\": \"GET\",\n \"operation_id\": \"list_blog_posts\",\n \"parameters\": [\n {\n \"description\": \"Use `published` or omit the parameter for public published posts. `all` and `draft` require authentication.\",\n \"in\": \"query\",\n \"name\": \"status\",\n \"required\": false,\n \"schema\": {\n \"default\": \"published\",\n \"enum\": [\n \"published\",\n \"draft\",\n \"all\"\n ],\n \"type\": \"string\"\n }\n }\n ],\n \"path\": \"/api/posts\",\n \"request_body\": null,\n \"security\": [\n {},\n {\n \"bearerAuth\": []\n },\n {\n \"apiKeyAuth\": []\n }\n ],\n \"skill_name\": \"list_blog_posts\",\n \"summary\": \"List posts\",\n \"tags\": [\n \"Posts\"\n ]\n },\n \"upsert_blog_post\": {\n \"description\": \"Upsert a post by slug\",\n \"destructive\": true,\n \"method\": \"PUT\",\n \"operation_id\": \"upsert_blog_post\",\n \"parameters\": [],\n \"path\": \"/api/posts/{slug}\",\n \"request_body\": {\n \"$ref\": \"#/components/schemas/BlogPostInput\"\n },\n \"security\": [\n {\n \"bearerAuth\": []\n },\n {\n \"apiKeyAuth\": []\n }\n ],\n \"skill_name\": \"upsert_blog_post\",\n \"summary\": \"Upsert a post by slug\",\n \"tags\": [\n \"Posts\"\n ]\n }\n}")
|
|
ROOT_SECURITY = json.loads("[]")
|
|
SECURITY_SCHEMES = json.loads("{\n \"apiKeyAuth\": {\n \"description\": \"Use the configured `BLOG_API_KEY` value.\",\n \"in\": \"header\",\n \"name\": \"x-api-key\",\n \"type\": \"apiKey\"\n },\n \"bearerAuth\": {\n \"description\": \"Use `Authorization: Bearer $BLOG_API_KEY`.\",\n \"scheme\": \"bearer\",\n \"type\": \"http\"\n }\n}")
|
|
SECURITY_FIELDS = json.loads("{\n \"apiKeyAuth\": {\n \"field\": \"BLOG_API_KEY\",\n \"kind\": \"apiKey\",\n \"location\": \"header\",\n \"name\": \"x-api-key\"\n },\n \"bearerAuth\": {\n \"field\": \"BLOG_API_KEY\",\n \"kind\": \"http\",\n \"scheme\": \"bearer\"\n }\n}")
|
|
|
|
|
|
class OperationInput(BaseModel):
|
|
parameters: dict[str, Any] = Field(
|
|
default_factory=dict,
|
|
description="Path, query, header, and cookie parameters keyed by OpenAPI parameter name.",
|
|
)
|
|
body: Any | None = Field(default=None, description="JSON request body, when the operation accepts one.")
|
|
|
|
|
|
class BlogOpenapiAgent(A2AAgent):
|
|
name = "blog-openapi-agent"
|
|
description = "Publishing and read API for the a2a cloud blog. Published content is public; draft, preview, create, update, and delete operations require the blog API key."
|
|
version = "0.1.0"
|
|
consumer_setup = ConsumerSetup.from_fields(
|
|
ConsumerSetupField.config("OPENAPI_BASE_URL", label="API base URL", description="Override the default API server (https://blog.a2acloud.io).", required=False, input_type="url"),
|
|
ConsumerSetupField.secret("BLOG_API_KEY", label="Blog API key", description="Used as either `Authorization: Bearer $BLOG_API_KEY` or the `x-api-key` header.", required=True),
|
|
)
|
|
llm_provisioning = LLMProvisioning.PLATFORM
|
|
pricing = Pricing(
|
|
price_per_call_usd=0.0,
|
|
caller_pays_llm=False,
|
|
notes="Uses a scoped platform LLM grant through ctx.llm.",
|
|
)
|
|
resources = Resources(cpu="200m", memory="512Mi")
|
|
egress = EgressPolicy(
|
|
allow_hosts=('blog.a2acloud.io',),
|
|
deny_internet_by_default=True,
|
|
)
|
|
tools_used = ("openapi", "deepagents")
|
|
capabilities = {
|
|
"openapi_auto_agent": {
|
|
"operation_count": len(OPERATIONS),
|
|
"default_base_url": DEFAULT_BASE_URL,
|
|
"security_schemes": list(SECURITY_SCHEMES),
|
|
}
|
|
}
|
|
|
|
@skill(
|
|
name="auto",
|
|
description="Use the OpenAPI service to complete a natural-language goal.",
|
|
tags=("openapi", "auto"),
|
|
timeout_seconds=180,
|
|
)
|
|
async def auto(self, ctx: RunContext, goal: str) -> dict[str, Any]:
|
|
creds = ctx.llm
|
|
if not creds.api_key:
|
|
return {
|
|
"error": "llm_credentials_missing",
|
|
"final": (
|
|
"LLM credentials were not available for this run. Hosted "
|
|
"generated OpenAPI agents should receive a platform LLM grant."
|
|
),
|
|
"messages": [],
|
|
}
|
|
model_kwargs: dict[str, Any] = {
|
|
"model": creds.model,
|
|
"base_url": creds.base_url,
|
|
"api_key": creds.api_key,
|
|
}
|
|
if creds.temperature_mode != "omit" and creds.temperature is not None:
|
|
model_kwargs["temperature"] = creds.temperature
|
|
if creds.extra_body:
|
|
model_kwargs["extra_body"] = dict(creds.extra_body)
|
|
model = ChatOpenAI(**model_kwargs)
|
|
graph = create_deep_agent(
|
|
model=model,
|
|
tools=self._operation_tools(ctx),
|
|
system_prompt=self._system_prompt(),
|
|
)
|
|
result = await graph.ainvoke({"messages": [{"role": "user", "content": goal}]})
|
|
messages = result.get("messages", []) if isinstance(result, dict) else []
|
|
final = _message_text(messages[-1]) if messages else result
|
|
return {
|
|
"final": final,
|
|
"messages": [_message_to_dict(message) for message in messages[-8:]],
|
|
}
|
|
|
|
@skill(
|
|
name="get_blog_health",
|
|
description="GET /api/healthz - Check blog service health",
|
|
tags=('Health',),
|
|
timeout_seconds=60,
|
|
)
|
|
async def get_blog_health(
|
|
self,
|
|
ctx: RunContext,
|
|
parameters: dict[str, Any] | None = None,
|
|
body: Any | None = None,
|
|
) -> dict[str, Any]:
|
|
return await self._request(
|
|
ctx,
|
|
"get_blog_health",
|
|
parameters=parameters or {},
|
|
body=body,
|
|
)
|
|
|
|
@skill(
|
|
name="list_blog_posts",
|
|
description="GET /api/posts - List posts",
|
|
tags=('Posts',),
|
|
timeout_seconds=60,
|
|
)
|
|
async def list_blog_posts(
|
|
self,
|
|
ctx: RunContext,
|
|
parameters: dict[str, Any] | None = None,
|
|
body: Any | None = None,
|
|
) -> dict[str, Any]:
|
|
return await self._request(
|
|
ctx,
|
|
"list_blog_posts",
|
|
parameters=parameters or {},
|
|
body=body,
|
|
)
|
|
|
|
@skill(
|
|
name="create_blog_post",
|
|
description="POST /api/posts - Create a blog post",
|
|
tags=('Posts',),
|
|
timeout_seconds=120,
|
|
)
|
|
async def create_blog_post(
|
|
self,
|
|
ctx: RunContext,
|
|
parameters: dict[str, Any] | None = None,
|
|
body: Any | None = None,
|
|
) -> dict[str, Any]:
|
|
return await self._request(
|
|
ctx,
|
|
"create_blog_post",
|
|
parameters=parameters or {},
|
|
body=body,
|
|
)
|
|
|
|
@skill(
|
|
name="get_blog_post",
|
|
description="GET /api/posts/{slug} - Get a post by slug",
|
|
tags=('Posts',),
|
|
timeout_seconds=60,
|
|
)
|
|
async def get_blog_post(
|
|
self,
|
|
ctx: RunContext,
|
|
parameters: dict[str, Any] | None = None,
|
|
body: Any | None = None,
|
|
) -> dict[str, Any]:
|
|
return await self._request(
|
|
ctx,
|
|
"get_blog_post",
|
|
parameters=parameters or {},
|
|
body=body,
|
|
)
|
|
|
|
@skill(
|
|
name="upsert_blog_post",
|
|
description="PUT /api/posts/{slug} - Upsert a post by slug",
|
|
tags=('Posts',),
|
|
timeout_seconds=120,
|
|
)
|
|
async def upsert_blog_post(
|
|
self,
|
|
ctx: RunContext,
|
|
parameters: dict[str, Any] | None = None,
|
|
body: Any | None = None,
|
|
) -> dict[str, Any]:
|
|
return await self._request(
|
|
ctx,
|
|
"upsert_blog_post",
|
|
parameters=parameters or {},
|
|
body=body,
|
|
)
|
|
|
|
@skill(
|
|
name="delete_blog_post",
|
|
description="DELETE /api/posts/{slug} - Delete a post by slug",
|
|
tags=('Posts',),
|
|
timeout_seconds=120,
|
|
)
|
|
async def delete_blog_post(
|
|
self,
|
|
ctx: RunContext,
|
|
parameters: dict[str, Any] | None = None,
|
|
body: Any | None = None,
|
|
) -> dict[str, Any]:
|
|
return await self._request(
|
|
ctx,
|
|
"delete_blog_post",
|
|
parameters=parameters or {},
|
|
body=body,
|
|
)
|
|
|
|
def _operation_tools(self, ctx: RunContext) -> list[StructuredTool]:
|
|
tools: list[StructuredTool] = []
|
|
for operation_id, operation in OPERATIONS.items():
|
|
async def call(
|
|
parameters: dict[str, Any] | None = None,
|
|
body: Any | None = None,
|
|
*,
|
|
_operation_id: str = operation_id,
|
|
) -> dict[str, Any]:
|
|
return await self._request(
|
|
ctx,
|
|
_operation_id,
|
|
parameters=parameters or {},
|
|
body=body,
|
|
)
|
|
|
|
tools.append(
|
|
StructuredTool.from_function(
|
|
coroutine=call,
|
|
name=operation["skill_name"],
|
|
description=self._tool_description(operation),
|
|
args_schema=OperationInput,
|
|
)
|
|
)
|
|
return tools
|
|
|
|
def _system_prompt(self) -> str:
|
|
lines = [
|
|
"You operate an API through generated OpenAPI tools.",
|
|
"Call tools to get real results. Do not invent API responses.",
|
|
"For write, update, or delete operations, explain the intended action before calling the tool.",
|
|
"If an operation reports missing consumer setup, tell the user which setup field is required.",
|
|
"",
|
|
"Available operations:",
|
|
]
|
|
for operation in OPERATIONS.values():
|
|
marker = "WRITE" if operation.get("destructive") else "READ"
|
|
lines.append(
|
|
f"- {operation['skill_name']}: {marker} {operation['method']} {operation['path']} — {operation['summary']}"
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
def _tool_description(self, operation: dict[str, Any]) -> str:
|
|
parts = [
|
|
f"{operation['method']} {operation['path']}",
|
|
str(operation.get("description") or operation.get("summary") or ""),
|
|
]
|
|
if operation.get("parameters"):
|
|
parts.append(
|
|
"Parameters: "
|
|
+ ", ".join(
|
|
f"{p['name']} in {p['in']}{' required' if p.get('required') else ''}"
|
|
for p in operation["parameters"]
|
|
)
|
|
)
|
|
if operation.get("request_body") is not None:
|
|
parts.append("Accepts a JSON request body.")
|
|
return "\n".join(part for part in parts if part)
|
|
|
|
async def _request(
|
|
self,
|
|
ctx: RunContext,
|
|
operation_id: str,
|
|
*,
|
|
parameters: dict[str, Any],
|
|
body: Any | None,
|
|
) -> dict[str, Any]:
|
|
operation = OPERATIONS[operation_id]
|
|
base_url = str(ctx.consumer_config("OPENAPI_BASE_URL", DEFAULT_BASE_URL) or DEFAULT_BASE_URL).rstrip("/")
|
|
url, query, headers = self._request_parts(ctx, operation, parameters)
|
|
request_kwargs: dict[str, Any] = {
|
|
"params": query,
|
|
"headers": headers,
|
|
}
|
|
if body is not None:
|
|
request_kwargs["json"] = body
|
|
async with httpx.AsyncClient(timeout=60.0, follow_redirects=True) as client:
|
|
response = await client.request(
|
|
operation["method"],
|
|
f"{base_url}{url}",
|
|
**request_kwargs,
|
|
)
|
|
content_type = response.headers.get("content-type", "")
|
|
try:
|
|
payload: Any = response.json() if "json" in content_type else response.text
|
|
except ValueError:
|
|
payload = response.text
|
|
if response.status_code >= 400:
|
|
return {
|
|
"ok": False,
|
|
"status_code": response.status_code,
|
|
"operation_id": operation_id,
|
|
"error": payload,
|
|
}
|
|
return {
|
|
"ok": True,
|
|
"status_code": response.status_code,
|
|
"operation_id": operation_id,
|
|
"result": payload,
|
|
}
|
|
|
|
def _request_parts(
|
|
self,
|
|
ctx: RunContext,
|
|
operation: dict[str, Any],
|
|
parameters: dict[str, Any],
|
|
) -> tuple[str, dict[str, Any], dict[str, str]]:
|
|
path = operation["path"]
|
|
query: dict[str, Any] = {}
|
|
headers: dict[str, str] = {}
|
|
cookies: dict[str, Any] = {}
|
|
for param in operation.get("parameters") or []:
|
|
name = param["name"]
|
|
if name not in parameters:
|
|
if param.get("required"):
|
|
raise ValueError(f"missing required parameter {name!r}")
|
|
continue
|
|
value = parameters[name]
|
|
location = param["in"]
|
|
if location == "path":
|
|
path = path.replace("{" + name + "}", str(value))
|
|
elif location == "query":
|
|
query[name] = value
|
|
elif location == "header":
|
|
headers[name] = str(value)
|
|
elif location == "cookie":
|
|
cookies[name] = value
|
|
if cookies:
|
|
headers["cookie"] = "; ".join(f"{key}={value}" for key, value in cookies.items())
|
|
auth_headers, auth_query = self._auth_for_operation(ctx, operation)
|
|
headers.update(auth_headers)
|
|
query.update(auth_query)
|
|
return path, query, headers
|
|
|
|
def _auth_for_operation(self, ctx: RunContext, operation: dict[str, Any]) -> tuple[dict[str, str], dict[str, Any]]:
|
|
requirements = operation.get("security")
|
|
if requirements is None:
|
|
requirements = ROOT_SECURITY
|
|
if requirements == []:
|
|
return {}, {}
|
|
missing: list[str] = []
|
|
for requirement in requirements or []:
|
|
if not isinstance(requirement, dict) or not requirement:
|
|
return {}, {}
|
|
headers: dict[str, str] = {}
|
|
query: dict[str, Any] = {}
|
|
ok = True
|
|
for scheme_name in requirement:
|
|
mapping = SECURITY_FIELDS.get(scheme_name) or {}
|
|
secret_name = mapping.get("field")
|
|
value = _consumer_secret_optional(ctx, secret_name) if secret_name else ""
|
|
if not value:
|
|
ok = False
|
|
if secret_name:
|
|
missing.append(secret_name)
|
|
continue
|
|
kind = mapping.get("kind")
|
|
if kind == "apiKey":
|
|
if mapping.get("location") == "query":
|
|
query[mapping.get("name") or scheme_name] = value
|
|
else:
|
|
headers[mapping.get("name") or scheme_name] = value
|
|
elif kind in {"http", "oauth2"}:
|
|
scheme = mapping.get("scheme") or "Bearer"
|
|
if scheme.lower() == "basic":
|
|
token = base64.b64encode(value.encode("utf-8")).decode("ascii")
|
|
headers["authorization"] = f"Basic {token}"
|
|
else:
|
|
headers["authorization"] = f"{scheme} {value}"
|
|
else:
|
|
ok = False
|
|
if ok:
|
|
return headers, query
|
|
if missing:
|
|
raise ConsumerSetupMissing(
|
|
"operation requires consumer setup secret(s): "
|
|
+ ", ".join(sorted(set(missing)))
|
|
)
|
|
return {}, {}
|
|
|
|
|
|
def _consumer_secret_optional(ctx: RunContext, name: str | None) -> str:
|
|
if not name:
|
|
return ""
|
|
try:
|
|
return ctx.consumer_secret(name)
|
|
except ConsumerSetupMissing:
|
|
return ""
|
|
|
|
|
|
def _message_text(message: Any) -> Any:
|
|
content = getattr(message, "content", message)
|
|
return content
|
|
|
|
|
|
def _message_to_dict(message: Any) -> dict[str, Any]:
|
|
if isinstance(message, BaseMessage):
|
|
return message.model_dump(mode="json")
|
|
if hasattr(message, "model_dump"):
|
|
return message.model_dump(mode="json")
|
|
if isinstance(message, dict):
|
|
return message
|
|
return {"content": str(message)}
|
|
|
|
|
|
agent = BlogOpenapiAgent()
|