Skip to content

Commit 39fbd71

Browse files
siacomuzzisiacomuzzi
committed
chore: auth0 sample
1 parent 17173ee commit 39fbd71

File tree

11 files changed

+1767
-340
lines changed

11 files changed

+1767
-340
lines changed

‎a2a-python-sdk/examples/async_user_confirmation/.env.example

+15
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
# Google Cloud
2+
GOOGLE_API_KEY=
3+
4+
# Auth0
5+
HR_AUTH0_DOMAIN=
6+
HR_AGENT_AUTH0_AUDIENCE=https://staff0/agent
7+
HR_API_AUTH0_AUDIENCE=https://staff0/
8+
9+
# A2A Client to HR AGENT client (auth method: Client Credentials)
10+
A2A_CLIENT_AUTH0_CLIENT_ID=
11+
A2A_CLIENT_AUTH0_CLIENT_SECRET=
12+
13+
# HR AGENT to HR API client (auth method: CIBA)
14+
HR_AGENT_AUTH0_CLIENT_ID=
15+
HR_AGENT_AUTH0_CLIENT_SECRET=

‎a2a-python-sdk/examples/async_user_confirmation/README.md

+65
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
This sample demonstrates how one of the agent's tools uses Auth0's Client-Initiated Backchannel Authentication (CIBA) feature to obtain tokens for accessing a separate API.
2+
Additionally, it shows agent-level authorization via the OAuth 2.0 Client Credentials flow.
3+
4+
## Prerequisites
5+
6+
- Python 3.12 or higher
7+
- [UV](https://docs.astral.sh/uv/)
8+
- [Gemini API key](https://ai.google.dev/gemini-api/docs/api-key)
9+
- An Auth0 tenant with the following configuration:
10+
- **APIs**
11+
- HR API
12+
- Audience: `https://staff0/`
13+
- Permissions: `read:employee`
14+
- HR Agent
15+
- Audience: `https://staff0/agent`
16+
- Permissions: `read:employee_status`
17+
- **Applications**
18+
- A2A Client
19+
- Grant Types: `Client Credentials`
20+
- APIs: `HR Agent` (enabled permissions: `read:employee_status`)
21+
- HR Agent
22+
- Grant Types: `Client Initiated Backchannel Authentication (CIBA)`
23+
- APIs: `Auth0 Management API` (enabled permissions: `read:users`)
24+
- Push Notifications using [Auth0 Guardian](https://auth0.com/docs/secure/multi-factor-authentication/auth0-guardian) must be `enabled`.
25+
- A test user enrolled in Guardian MFA.
26+
27+
## Getting started
28+
29+
1. Start the server
30+
31+
```bash
32+
uv run --prerelease=allow .
33+
```
34+
35+
2. Run the test client
36+
```bash
37+
uv run --prerelease=allow test_client.py
38+
```
39+
40+
## How it works
41+
42+
Allows an A2A client to interact with an external HR Agent, owned by Staff0, to verify whether the provided user data corresponds to an active Staff0 employee.
43+
With the user’s authorization (via push notification), the Staff0 HR Agent can access the internal company HR API to retrieve their employment details.
44+
45+
```mermaid
46+
sequenceDiagram
47+
participant User as John Doe
48+
participant A2AClient as A2A Client
49+
participant Auth0 as Auth0 (staff0.auth0.com)
50+
participant HR_Agent as Staff0 HR Agent
51+
participant HR_API as Staff0 HR API
52+
53+
A2AClient->>HR_Agent: Get A2A Agent Card
54+
HR_Agent-->>A2AClient: Agent Card
55+
A2AClient->>Auth0: Request access token (client credentials)
56+
Auth0-->>A2AClient: Access Token
57+
A2AClient->>HR_Agent: Does John Doe with email [email protected] work at Staff0? (Access Token)
58+
HR_Agent->>Auth0: Request access token (CIBA)
59+
Auth0->>User: Push notification to approve access
60+
User-->>Auth0: Approves access
61+
Auth0-->>HR_Agent: Access Token
62+
HR_Agent->>HR_API: Retrieve employment details (Access Token)
63+
HR_API-->>HR_Agent: Employment details
64+
HR_Agent-->>A2AClient: Yes, John Doe is an active employee.
65+
```

‎a2a-python-sdk/examples/async_user_confirmation/__main__.py

+113
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,113 @@
1+
import asyncio
2+
import json
3+
import logging
4+
import sys
5+
import os
6+
import click
7+
8+
from dotenv import load_dotenv
9+
import uvicorn
10+
load_dotenv()
11+
12+
from agent import HRAgent
13+
from agent_executor import HRAgentExecutor
14+
from oauth2_middleware import OAuth2Middleware
15+
from api import hr_api
16+
17+
from a2a.server.apps import A2AStarletteApplication
18+
from a2a.server.request_handlers import DefaultRequestHandler
19+
from a2a.server.tasks import InMemoryTaskStore
20+
from a2a.types import (
21+
AgentAuthentication,
22+
AgentCapabilities,
23+
AgentCard,
24+
AgentSkill,
25+
# ClientCredentialsOAuthFlow,
26+
# OAuth2SecurityScheme,
27+
# OAuthFlows,
28+
)
29+
30+
31+
logging.basicConfig(level=logging.INFO)
32+
logger = logging.getLogger()
33+
34+
35+
@click.command()
36+
@click.option('--host', default='0.0.0.0')
37+
@click.option('--port_agent', default=10050)
38+
@click.option('--port_api', default=10051)
39+
def main(host: str, port_agent: int, port_api: int):
40+
async def run_all():
41+
await asyncio.gather(
42+
start_agent(host, port_agent),
43+
start_api(host, port_api),
44+
)
45+
46+
asyncio.run(run_all())
47+
48+
49+
async def start_agent(host: str, port):
50+
agent_card = AgentCard(
51+
name='Staff0 HR Agent',
52+
description='This agent handles external verification requests about Staff0 employees made by third parties.',
53+
url=f'http://{host}:{port}/',
54+
version='1.0.0',
55+
defaultInputModes=HRAgent.SUPPORTED_CONTENT_TYPES,
56+
defaultOutputModes=HRAgent.SUPPORTED_CONTENT_TYPES,
57+
capabilities=AgentCapabilities(streaming=True),
58+
skills=[
59+
AgentSkill(
60+
id='is_active_employee',
61+
name='Check Employment Status Tool',
62+
description='Confirm whether a person is an active employee of the company.',
63+
tags=['employment status'],
64+
examples=[
65+
'Does John Doe with email [email protected] work at Staff0?'
66+
],
67+
)
68+
],
69+
authentication=AgentAuthentication(schemes=['public']),
70+
# securitySchemes={
71+
# 'oauth2_m2m_client': OAuth2SecurityScheme(
72+
# description='',
73+
# flows=OAuthFlows(
74+
# authorizationCode=ClientCredentialsOAuthFlow(
75+
# tokenUrl=f'https://{os.getenv("HR_AUTH0_DOMAIN")}/oauth/token',
76+
# scopes={
77+
# 'read:employee_status': 'Allows confirming whether a person is an active employee of the company.',
78+
# },
79+
# ),
80+
# ),
81+
# ),
82+
# },
83+
# security=[{
84+
# 'oauth2_m2m_client': [
85+
# 'read:employee_status',
86+
# ],
87+
# }],
88+
)
89+
90+
request_handler = DefaultRequestHandler(
91+
agent_executor=HRAgentExecutor(),
92+
task_store=InMemoryTaskStore(),
93+
)
94+
95+
server = A2AStarletteApplication(
96+
agent_card=agent_card, http_handler=request_handler
97+
)
98+
99+
app = server.build()
100+
# app.add_middleware(OAuth2Middleware, agent_card=agent_card, public_paths=['/.well-known/agent.json'])
101+
102+
logger.info(f'Starting HR Agent server on {host}:{port}')
103+
await uvicorn.Server(uvicorn.Config(app=app, host=host, port=port)).serve()
104+
105+
106+
async def start_api(host: str, port):
107+
logger.info(f'Starting HR API server on {host}:{port}')
108+
await uvicorn.Server(uvicorn.Config(app=hr_api, host=host, port=port)).serve()
109+
110+
111+
# this ensures that `main()` runs when using `uv run .`
112+
if not hasattr(sys, "_called_from_uvicorn"):
113+
main()

0 commit comments

Comments
 (0)