bobseares-2/learnhouse-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

deploy

Browse Source
This commit is contained in:
a2a-platform
2026-06-29 00:30:43 +00:00
parent ae71590f72
commit c8cea595f2
4 changed files with 49 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
.a2a/agent.dsl.json
View File

@@ -3,7 +3,7 @@
"language": "python",
"name": "learnhouse-agent",
"description": "LearnHouse is an open-source platform tailored for learning experiences.",
"version": "1.2.7",
"version": "1.2.8",
"entrypoint": {
"module": "agent",
"class_name": "Learnhouse",
@@ -88,15 +88,6 @@
"required": false,
"input_type": "url",
"options": []
},
{
"name": "LEARNHOUSE_API_KEY",
"kind": "secret",
"label": "LearnHouse API key",
"description": "Bearer token or API key for LearnHouse requests.",
"required": true,
"input_type": "password",
"options": []
}
]
},
@@ -141,13 +132,10 @@
"meta_agent_manifest": null,
"state_schema": null,
"workspace_access": {
"enabled": true,
"max_files": 50,
"allowed_modes": [
"read_only",
"read_write_overlay"
],
"require_reason": false,
"enabled": false,
"max_files": 0,
"allowed_modes": [],
"require_reason": true,
"deny_patterns": [],
"require_human_approval": false,
"max_total_size_bytes": 104857600
@@ -175,7 +163,7 @@
"source": "python-a2a-pack",
"project_manifest": {
"name": "learnhouse-agent",
"version": "1.2.7",
"version": "1.2.8",
"entrypoint": "agent:Learnhouse"
}
}

8
README.md
View File

@@ -7,5 +7,13 @@ Generated A2APack agent for LearnHouse.
- Main skill: `auto`
- Routing mode: DeepAgents subagents across 35 route groups
Deployment context:
- This LearnHouse instance runs single-organization tenancy.
- Existing organization: A2A Cloud Academy
- Default `org_slug`: `default`
- Do not create additional organizations on this instance; that requires
LearnHouse Enterprise multi-organization mode.
The generated code is intentionally editable. Tune prompts, operation
grouping, auth names, and safety policy before publishing serious agents.

10
a2a.yaml
View File

@@ -1,5 +1,5 @@
name: learnhouse-agent
version: 1.2.7
version: 1.2.8
entrypoint: agent:Learnhouse
description: LearnHouse is an open-source platform tailored for learning experiences.
runtime:
@@ -10,11 +10,3 @@ runtime:
allow_hosts:
- learnhouse.a2acloud.io
deny_internet_by_default: true
consumer_setup:
fields:
- name: LEARNHOUSE_API_KEY
kind: secret
label: LearnHouse API key
description: Bearer token or API key for LearnHouse requests.
required: true
input_type: password

151
agent.py
View File

@@ -1,7 +1,6 @@
from __future__ import annotations
import base64
import hashlib
import json
import re
from pathlib import Path
@@ -23,8 +22,6 @@ from a2a_pack import (
Pricing,
Resources,
RunContext,
WorkspaceAccess,
WorkspaceMode,
skill,
)
@@ -39,34 +36,15 @@ PATH_PARAMETER_RE = re.compile(r"{([^}/]+)}")
SOURCE_ROOT = Path(globals().get("__file__", "agent.py")).resolve().parent
SOURCE_SKILLS_DIR = SOURCE_ROOT / "skills"
RUNTIME_SKILLS_DIR = ".deepagents/openapi-skills/"
MAX_OPENAI_TOOL_NAME_LENGTH = 64
def _openai_safe_tool_name(name: str, used: set[str]) -> str:
safe = re.sub(r"[^a-zA-Z0-9_-]+", "_", name).strip("_") or "operation"
if len(safe) > MAX_OPENAI_TOOL_NAME_LENGTH:
digest = hashlib.sha1(safe.encode("utf-8")).hexdigest()[:10]
prefix = safe[: MAX_OPENAI_TOOL_NAME_LENGTH - len(digest) - 1].rstrip("_-")
safe = f"{prefix or 'operation'}_{digest}"
candidate = safe
index = 2
while candidate in used:
suffix = f"_{index}"
candidate = safe[: MAX_OPENAI_TOOL_NAME_LENGTH - len(suffix)].rstrip("_-") + suffix
index += 1
used.add(candidate)
return candidate
def _normalize_operation_tool_names() -> None:
used: set[str] = set()
for operation_id, operation in OPERATIONS.items():
original = str(operation.get("skill_name") or operation_id)
operation["original_skill_name"] = original
operation["skill_name"] = _openai_safe_tool_name(original, used)
_normalize_operation_tool_names()
LEARNHOUSE_CONTEXT = (
"Deployment context for this LearnHouse instance: it runs single-organization "
"tenancy at https://learnhouse.a2acloud.io. The existing organization is "
"A2A Cloud Academy with org_slug 'default'. Do not try to create additional "
"organizations on this instance; multi-organization mode requires LearnHouse "
"Enterprise. When an operation needs org_slug and the user has not supplied "
"one, use 'default'. For list operations with required path page/limit and no "
"user-supplied values, use page=1 and limit=50."
)
class OperationInput(BaseModel):
@@ -80,10 +58,9 @@ class OperationInput(BaseModel):
class Learnhouse(A2AAgent):
name = "learnhouse"
description = "LearnHouse is an open-source platform tailored for learning experiences."
version = "1.2.7"
version = "1.2.8"
consumer_setup = ConsumerSetup.from_fields(
ConsumerSetupField.config("OPENAPI_BASE_URL", label="API base URL", description="Override the default API server (https://learnhouse.a2acloud.io).", required=False, input_type="url"),
ConsumerSetupField.secret("LEARNHOUSE_API_KEY", label="LearnHouse API key", description="Bearer token or API key for LearnHouse requests.", required=True),
)
llm_provisioning = LLMProvisioning.PLATFORM
pricing = Pricing(
@@ -92,11 +69,6 @@ class Learnhouse(A2AAgent):
notes="Uses the caller's saved LLM credential through ctx.llm.",
)
resources = Resources(cpu="200m", memory="512Mi")
workspace_access = WorkspaceAccess.dynamic(
max_files=50,
allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY),
require_reason=False,
)
egress = EgressPolicy(
allow_hosts=('learnhouse.a2acloud.io',),
deny_internet_by_default=True,
@@ -121,9 +93,6 @@ class Learnhouse(A2AAgent):
timeout_seconds=900,
)
async def auto(self, ctx: RunContext, goal: str) -> dict[str, Any]:
smoke_result = await self._maybe_smoke_test(ctx, goal)
if smoke_result is not None:
return smoke_result
creds = ctx.llm
if not creds.api_key:
return {
@@ -154,23 +123,6 @@ class Learnhouse(A2AAgent):
async def _maybe_smoke_test(self, ctx: RunContext, goal: str) -> dict[str, Any] | None:
operation_id = _select_smoke_operation(goal)
if operation_id is None:
return None
operation = OPERATIONS[operation_id]
result = await self._request(ctx, operation_id, parameters={}, body=None)
body = result.get("result") if result.get("ok") else result.get("error")
final = {
"endpoint_called": str(operation.get("method")) + " " + str(operation.get("path")),
"http_status": result.get("status_code"),
"body_preview": _body_preview(body, 200),
}
return {
"final": json.dumps(final, indent=2, ensure_ascii=False),
"messages": [],
}
def _operation_subagents(self, ctx: RunContext, skills_root: str) -> list[dict[str, Any]]:
subagents: list[dict[str, Any]] = []
for group in OPERATION_GROUPS:
@@ -226,9 +178,9 @@ class Learnhouse(A2AAgent):
if OPERATION_GROUPS:
lines = [
"You coordinate route-specific OpenAPI subagents.",
LEARNHOUSE_CONTEXT,
"Delegate API work to the matching subagent with the task tool.",
"Do not invent API responses; ask subagents to call their tools for real results.",
"For smoke tests, health checks, status checks, or ping-style requests, prefer explicit health/status/ping route groups over a generic root '/' route group.",
"If a subagent reports missing consumer setup, tell the user which setup field is required.",
"If a subagent reports that the generated OpenAPI agent may be stale, ask whether the user wants to refresh it from the latest OpenAPI spec.",
"",
@@ -241,9 +193,8 @@ class Learnhouse(A2AAgent):
return "\n".join(lines)
lines = [
"You operate an API through generated OpenAPI tools.",
LEARNHOUSE_CONTEXT,
"Call tools to get real results. Do not invent API responses.",
"For smoke tests, health checks, status checks, or ping-style requests, prefer explicit health/status/ping GET operations over generic root '/' operations.",
"Use a root '/' operation only when the user asks for the service root/homepage or no more specific read-only operation fits.",
"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.",
"If an operation returns 404, 405, 410, or a schema/validation error that suggests the live API no longer matches these tools, tell the user this generated agent may need to be refreshed from the latest OpenAPI spec and ask whether they want to refresh it.",
@@ -260,10 +211,9 @@ class Learnhouse(A2AAgent):
def _subagent_prompt(self, group: dict[str, Any]) -> str:
lines = [
"You operate one route group from a generated OpenAPI API.",
LEARNHOUSE_CONTEXT,
"Use the available operation tools to make real API calls. Do not invent API responses.",
"Read the route group's SKILL.md when it applies; it lists every route you cover.",
"For smoke tests, health checks, status checks, or ping-style requests, prefer explicit health/status/ping GET operations over generic root '/' operations.",
"Use a root '/' operation only when the user asks for the service root/homepage or no more specific read-only operation fits.",
"For write, update, or delete operations, explain the intended action before calling the tool.",
"If an operation reports missing consumer setup, return the exact setup field name.",
"If an operation returns 404, 405, 410, or a schema/validation error that suggests the live API no longer matches these tools, say the generated agent may need to be refreshed from the latest OpenAPI spec.",
@@ -295,7 +245,6 @@ class Learnhouse(A2AAgent):
)
if operation.get("request_body") is not None:
parts.append("Accepts a JSON request body.")
parts.append(f"Operation ID: {operation.get('operation_id') or operation.get('original_skill_name')}")
return "\n".join(part for part in parts if part)
async def _request(
@@ -310,6 +259,7 @@ class Learnhouse(A2AAgent):
default_base_url = str(operation.get("base_url") or DEFAULT_BASE_URL).rstrip("/")
base_url_field = str(operation.get("base_url_field") or "OPENAPI_BASE_URL")
base_url = str(ctx.consumer_config(base_url_field, default_base_url) or default_base_url).rstrip("/")
parameters = _apply_learnhouse_defaults(operation, parameters)
url, query, headers = self._request_parts(ctx, operation, parameters)
request_kwargs: dict[str, Any] = {
"params": query,
@@ -342,7 +292,6 @@ class Learnhouse(A2AAgent):
"ok": False,
"status_code": response.status_code,
"operation_id": operation_id,
"tool_name": operation.get("skill_name"),
"error": payload,
}
if response.status_code in {404, 405, 410, 422}:
@@ -355,7 +304,6 @@ class Learnhouse(A2AAgent):
"ok": True,
"status_code": response.status_code,
"operation_id": operation_id,
"tool_name": operation.get("skill_name"),
"result": payload,
}
@@ -445,13 +393,6 @@ class Learnhouse(A2AAgent):
return {}, {}
def _learnhouse_auth_header(value: str) -> str:
token = value.strip()
if token.lower().startswith(("bearer ", "basic ", "token ")):
return token
return f"Bearer {token}"
def _consumer_secret_optional(ctx: RunContext, name: str | None) -> str:
if not name:
return ""
@@ -461,50 +402,26 @@ def _consumer_secret_optional(ctx: RunContext, name: str | None) -> str:
return ""
def _select_smoke_operation(goal: str) -> str | None:
text = str(goal or "").lower()
smoke_terms = ("smoke", "health", "status", "ping", "readiness", "liveness")
if not any(term in text for term in smoke_terms):
return None
preferred_terms = ("health", "status", "ping", "ready", "readiness", "live", "liveness")
candidates: list[tuple[int, int, str]] = []
for operation_id, operation in OPERATIONS.items():
if str(operation.get("method") or "").upper() != "GET":
continue
if operation.get("destructive"):
continue
if any(param.get("required") for param in operation.get("parameters") or []):
continue
path = str(operation.get("path") or "")
searchable = " ".join(
str(operation.get(key) or "")
for key in ("operation_id", "skill_name", "path", "summary", "description")
).lower()
is_root = path.strip() == "/"
matched_preferred = next(
(index for index, term in enumerate(preferred_terms) if term in searchable),
None,
)
if matched_preferred is not None:
score = matched_preferred
elif "smoke" in text and not is_root:
score = 100
elif "smoke" in text:
score = 1000
else:
continue
candidates.append((score, len(path), operation_id))
if not candidates:
return None
return min(candidates)[2]
def _body_preview(body: Any, limit: int) -> str:
if isinstance(body, str):
text = body
else:
text = json.dumps(body, ensure_ascii=False)
return text[:limit]
def _apply_learnhouse_defaults(
operation: dict[str, Any],
parameters: dict[str, Any],
) -> dict[str, Any]:
params = dict(parameters)
required_path_names = {
str(param.get("name"))
for param in operation.get("parameters") or []
if param.get("required") and param.get("in") == "path"
}
if "org_slug" in required_path_names and not params.get("org_slug"):
params["org_slug"] = "default"
method = str(operation.get("method") or "").upper()
summary = str(operation.get("summary") or "").lower()
if method == "GET" and "list" in summary:
if "page" in required_path_names and not params.get("page"):
params["page"] = 1
if "limit" in required_path_names and not params.get("limit"):
params["limit"] = 50
return params
def _runtime_skills_root(ctx: RunContext) -> str: