Metadata-Version: 2.3
Name: spakky-a2a
Version: 6.12.0
Summary: A2A (Agent2Agent) protocol server plugin for Spakky framework
Author: Spakky
Author-email: Spakky <sejong418@icloud.com>
License: MIT
Requires-Dist: a2a-sdk[http-server]>=1.1.0
Requires-Dist: grpcio>=1.68.0
Requires-Dist: pydantic>=2.4
Requires-Dist: pydantic-settings>=2.13.1
Requires-Dist: spakky>=6.12.0
Requires-Dist: spakky-agent>=6.12.0
Requires-Python: >=3.12
Description-Content-Type: text/markdown

# spakky-a2a

> `spakky-a2a`는 `spakky-agent`를 A2A (Agent2Agent) protocol server와 원격 teammate delegation으로 노출하는 adapter plugin입니다.
> Spakky `@Agent`를 A2A server로 공개하고, agent spec/tool catalog/teammates에서 AgentCard를 유도하며, 공식 `a2a-sdk` client 위에 core `IAgentDelegate` port를 구현합니다.

## 설치

```bash
pip install spakky-a2a
```

실행 가능한 agent에는 별도 `IAgentModel` provider가 필요합니다. Durable run 또는 HITL resume을 사용하면 `spakky-sqlalchemy[agent]` 같은 provider가 공급하는 `spakky-agent` persistence repository도 필요합니다.

## 설정

`A2AConfig`는 `SPAKKY_A2A_` 접두사의 환경변수를 읽습니다.

| 환경변수 | 기본값 | 목적 |
|----------------------|---------|---------|
| `SPAKKY_A2A_DEFAULT_BASE_URL` | `http://localhost:8000` | marker가 `base_url`을 생략할 때 mount path와 결합할 public host URL |
| `SPAKKY_A2A_DEFAULT_VERSION` | `1.0.0` | derived AgentCard에 광고할 semantic version |
| `SPAKKY_A2A_DEFAULT_MOUNT_PATH_PREFIX` | `/a2a` | 자동 mount되는 A2A agent app의 path prefix |

Plugin 초기화는 `A2AConfig`, `A2AAgentRegistry`, `A2AAgentServerSpec`, A2A remote delegate Pod, 그리고 `@Agent`와 `@A2ACompatible`가 함께 붙은 class를 발견해 ASGI/gRPC host에 연결하는 post-processor를 등록합니다.

## Agent 노출

`@A2ACompatible`는 `@Agent`와 같은 class에 쌓는 tag입니다. `@Agent`가 Pod를 등록하고, tag는 A2A transport metadata와 optional mount path를 기록합니다.

```python
from spakky.agent import Agent, AgentExecutionSpec, IAgentModel
from spakky.plugins.a2a import A2ACompatible


@A2ACompatible(
    base_url="https://agents.example.com/a2a/planner",
    version="1.0.0",
    mount_path="/a2a/planner",
    rest_mount_path="/a2a-rest/planner",
    rest_base_url="https://agents.example.com/a2a-rest/planner",
    grpc_enabled=True,
    grpc_base_url="grpc://agents.example.com:443",
)
@Agent(spec=AgentExecutionSpec(name="planner", objective="Plan work"))
class PlannerAgent:
    def __init__(self, model: IAgentModel) -> None:
        self.model = model
```

애플리케이션이 Starlette/FastAPI host Pod를 제공하면 plugin post-processor가 bootstrap 중
`mount_path`에 A2A JSON-RPC + AgentCard app을 자동 mount합니다. `rest_mount_path`를 지정하면
HTTP+JSON REST + AgentCard app도 그 path에 별도로 mount합니다. `mount_path`를 생략하면
`{default_mount_path_prefix}/{agent_name}`을 사용합니다. `version`을 생략하면
`A2AConfig.default_version`을 사용합니다.

`base_url`은 AgentCard `supported_interfaces[].url`에 광고되는 **public transport endpoint**입니다.
클라이언트가 reverse proxy나 ASGI mount path를 통해 호출한다면 그 외부 path를 포함해야 하며,
`/.well-known/agent-card.json` 자체는 포함하지 않습니다. 예를 들어 위 선언에서 card URL은
`https://agents.example.com/a2a/planner/.well-known/agent-card.json`이고, JSON-RPC operation endpoint는
`https://agents.example.com/a2a/planner/`입니다. `base_url`을 생략하면 framework가
`A2AConfig.default_base_url.rstrip("/") + mount_path`로 유도하므로 기본 설정에서는
`http://localhost:8000/a2a/planner`가 광고됩니다. REST도 `rest_base_url`을 생략하면
`default_base_url + rest_mount_path`를 광고합니다.

```python
from starlette.applications import Starlette
from spakky.core.pod.annotations.pod import Pod


@Pod(name="asgi_host")
def asgi_host() -> Starlette:
    return Starlette()
```

gRPC 노출은 `grpc_enabled=True`인 entry를 `spakky-grpc`의 `GrpcServerSpec`에 자동 등록합니다.
따라서 애플리케이션은 `spakky-a2a`와 `spakky-grpc` plugin을 함께 로드하고
`SPAKKY_GRPC_BIND_ADDRESSES`를 설정하면 됩니다. `spakky-grpc`가 로드되지 않은 애플리케이션에서는
gRPC 선언은 no-op입니다.

애플리케이션 bootstrap 이후 `A2AAgentServerSpec.build_app_for("planner")`,
`build_rest_app_for("planner")`, `build_grpc_handler_for("planner")`는 특수 host나 테스트에서 쓰는
lower-level escape hatch입니다. 일반 애플리케이션은 `@A2ACompatible` 선언과 host Pod만 사용합니다.

## AgentCard 유도

`AgentCardFactory`는 다음 입력에서 card를 유도합니다.

- name/description: `AgentExecutionSpec.name`, `objective`, 또는 `instructions`
- streaming capability: `streaming_exposure_mode`; `NO_STREAM_UNTIL_FINAL_GUARDED`는 streaming capability 노출을 끕니다.
- tool: synthetic teammate delegation tool을 제외한 native `@agent_tool` descriptor
- delegation skill: 선언된 `AgentTeammate` entry

## 원격 Teammate 위임

`A2AAgentDelegate`는 `AgentExecutionSpec.teammates` entry가 원격 AgentCard URL을 가리키는 teammate를 위해 core `IAgentDelegate` port를 구현합니다. Plugin 초기화가 `A2AAgentDelegate`를 Pod로 등록하고 `IAgentDelegate`에 바인딩하므로 parent agent는 `IAgentDelegate` 또는 `A2AAgentDelegate`를 생성자 주입으로 받을 수 있습니다. Core agent runner는 각 teammate를 `teammate.<schema_token(name)>.delegate`라는 model-callable delegation tool로 노출합니다. `schema_token`은 teammate name의 앞뒤 공백을 제거한 뒤 `[a-zA-Z0-9_]`가 아닌 연속 문자를 단일 `_`로 치환하고, 앞뒤 `_`를 제거한 다음 소문자화한 값입니다. 이 결과가 비면 agent definition 단계에서 거부됩니다. Local teammate Pod는 in-process로 실행하고, remote teammate는 공식 `a2a-sdk` client를 사용합니다.

```python
from spakky.agent import Agent, AgentExecutionSpec, AgentTeammate
from spakky.plugins.a2a import A2AAgentDelegate


@Agent(
    spec=AgentExecutionSpec(
        name="orchestrator",
        teammates=(
            AgentTeammate(
                name="researcher",
                card_url="https://agents.example.com/.well-known/agent-card.json",
            ),
        ),
    )
)
class Orchestrator:
    def __init__(self, delegate: A2AAgentDelegate) -> None:
        self.delegate = delegate
```

원격 delegation은 SDK client로 `message/send`를 보내고 remote task stream을 추적한 뒤, child task/message/artifact update를 parent run id를 보존한 Spakky protocol-neutral event stream으로 되돌립니다.

## REST HTTP+JSON Transport

SDK route 이름은 JSON-RPC method 문자열과 다릅니다.

| A2A operation | REST route |
|---------------|------------|
| `message/send` | `POST /message:send` |
| `message/stream` | `POST /message:stream` |
| `tasks/get` | `GET /tasks/{id}` |
| `tasks/cancel` | `POST /tasks/{id}:cancel` |
| `tasks/subscribe` | `GET /tasks/{id}:subscribe` or `POST /tasks/{id}:subscribe` |

REST request/response body는 A2A SDK protobuf JSON encoding을 사용합니다. 예를 들어 user message는 `{"message":{"role":"ROLE_USER","messageId":"m1","parts":[{"text":"hi"}]}}` 형태로 보냅니다.

## HITL와 Auth Interrupt

`SpakkyAgentExecutor`는 core `AgentRunner.run_events()` stream을 소비합니다. Approval/auth pause는 successful terminal `RunFinishedEvent`가 아니라 protocol-neutral `RunPausedEvent`로 들어옵니다. A2A projector는 `reason=approval_required`를 `TASK_STATE_INPUT_REQUIRED`로 매핑하고 approval id와 allowed decisions를 data part에 포함합니다. `reason=auth_required`는 `TASK_STATE_AUTH_REQUIRED`로 매핑하므로, run stream이 끝난 뒤 durable `state.reason`을 다시 조회하지 않아도 auth-required 상태를 표현할 수 있습니다.

Approval resume은 `approval_id`와 `decision`을 가진 inbound A2A data part로 전달됩니다. Executor는 `APPROVAL_DECISION` signal을 append하고 같은 task id로 `RunAgentInput(resume=True)`를 다시 실행합니다.

## Task Store

Server transport는 synchronous `IA2ATaskRepository` port 위에 async `a2a-sdk` `TaskStore`를 얹는 bridge인 `SpakkyA2ATaskStore`를 사용합니다. Builder 인자로 repository를 주지 않고 container에도 repository Pod가 없으면 plugin은 `InMemoryA2ATaskRepository`를 사용합니다.

## 개발 검증

패키지 단위 검증은 해당 패키지 디렉토리에서 실행합니다.

```bash
uv run ruff format .
uv run ruff check .
uv run pyrefly check
uv run pytest
```

`pytest`는 각 패키지 `pyproject.toml`의 coverage 설정을 사용합니다.

## 라이선스

MIT License
