a2a-source-edit: write agent.py

This commit is contained in:
a2a-cloud
2026-07-14 21:41:42 +00:00
parent c73f2db1ec
commit eb7c5e8a63

View File

@@ -306,27 +306,13 @@ class SitsBetweenParentsChaotic19Aab317(A2AAgent[SitsBetweenParentsChaotic19Aab3
creds: LLMCreds, creds: LLMCreds,
) -> Any: ) -> Any:
from a2a_pack.deepagents import create_a2a_deep_agent from a2a_pack.deepagents import create_a2a_deep_agent
from langchain_core.tools import tool from langchain_core.tools import StructuredTool
@tool
def validate_parent_plan_json(plan_json: str) -> str:
"""Validate the JSON plan shape and flag missing parent-planning sections."""
try:
data = json.loads(plan_json)
except Exception as exc: # noqa: BLE001
return json.dumps({"valid": False, "error": f"invalid_json: {exc}"})
required = [
"answer",
"school_tomorrow",
"checklist",
"calendar_reminders",
"payment_reminders",
"backpack_alerts",
"morning_summaries",
]
missing = [key for key in required if key not in data]
return json.dumps({"valid": not missing, "missing": missing}, ensure_ascii=False)
validation_tool = StructuredTool.from_function(
func=_validate_parent_plan_json,
name="validate_parent_plan_json",
description="Validate the JSON plan shape and flag missing parent-planning sections.",
)
backend = ctx.workspace_backend() backend = ctx.workspace_backend()
skill_sources = _seed_runtime_skills(backend, ctx) skill_sources = _seed_runtime_skills(backend, ctx)
return create_a2a_deep_agent( return create_a2a_deep_agent(
@@ -334,7 +320,7 @@ class SitsBetweenParentsChaotic19Aab317(A2AAgent[SitsBetweenParentsChaotic19Aab3
creds=creds, creds=creds,
backend=backend, backend=backend,
skills=skill_sources or None, skills=skill_sources or None,
tools=[validate_parent_plan_json], tools=[validation_tool],
system_prompt=SYSTEM_PROMPT, system_prompt=SYSTEM_PROMPT,
) )
@@ -425,6 +411,24 @@ class SitsBetweenParentsChaotic19Aab317(A2AAgent[SitsBetweenParentsChaotic19Aab3
return previews return previews
def _validate_parent_plan_json(plan_json: str) -> str:
try:
data = json.loads(plan_json)
except Exception as exc: # noqa: BLE001
return json.dumps({"valid": False, "error": f"invalid_json: {exc}"})
required = [
"answer",
"school_tomorrow",
"checklist",
"calendar_reminders",
"payment_reminders",
"backpack_alerts",
"morning_summaries",
]
missing = [key for key in required if key not in data]
return json.dumps({"valid": not missing, "missing": missing}, ensure_ascii=False)
def _runtime_skills_root(ctx: RunContext[Any]) -> str: def _runtime_skills_root(ctx: RunContext[Any]) -> str:
workspace = getattr(ctx, "_workspace", None) workspace = getattr(ctx, "_workspace", None)
prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ()) prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())