Skip to content

Commit 56be9da

Browse files
siacomuzzisiacomuzzi
committed
chore: auth0 sample (Auth for Headless Agents)
1 parent 8a44cff commit 56be9da

File tree

10 files changed

+2265
-0
lines changed

10 files changed

+2265
-0
lines changed

‎samples/python/agents/headless_agent_auth/.env.example

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

‎samples/python/agents/headless_agent_auth/README.md

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

‎samples/python/agents/headless_agent_auth/__main__.py

+113
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,113 @@
1+
import asyncio
2+
import click
3+
# import json
4+
import logging
5+
# import os
6+
import sys
7+
import uvicorn
8+
9+
from dotenv import load_dotenv
10+
load_dotenv()
11+
12+
from agent import HRAgent
13+
from agent_executor import HRAgentExecutor
14+
from api import hr_api
15+
# from oauth2_middleware import OAuth2Middleware
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='0.1.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)