A robust, enterprise-grade Python library for Traylinx Sentinel Agent-to-Agent (A2A) authentication. This client provides secure token management, automatic retry logic, comprehensive error handling, and seamless integration with FastAPI applications.
- π Dual Token Authentication: Handles both
access_tokenandagent_secret_tokenwith automatic refresh - π‘οΈ Enterprise Security: Input validation, secure credential handling, and comprehensive error management
- β‘ High Performance: Connection pooling, automatic retries with exponential backoff, and efficient token caching
- π Thread-Safe: Built-in thread safety for concurrent applications and production environments
- π― FastAPI Integration: Simple decorators for protecting endpoints with A2A authentication
- π‘ JSON-RPC Support: Full support for A2A RPC method calls with automatic credential detection
- π§ Zero Configuration: Works with environment variables out of the box
- π Production Ready: Configurable logging, monitoring, and comprehensive error handling
pip install traylinx-auth-clientpoetry add traylinx-auth-client- Python 3.8 or higher
requests>= 2.25.0pydantic>= 2.0.0
from traylinx_auth_client import make_a2a_request
# Set environment variables: TRAYLINX_CLIENT_ID, TRAYLINX_CLIENT_SECRET,
# TRAYLINX_API_BASE_URL, TRAYLINX_AGENT_USER_ID
# Make authenticated request to another agent
response = make_a2a_request("GET", "https://other-agent.com/api/data")
print(response) # JSON response from the agentSet these environment variables for your agent:
export TRAYLINX_CLIENT_ID="your-client-id"
export TRAYLINX_CLIENT_SECRET="your-client-secret"
export TRAYLINX_API_BASE_URL="https://auth.traylinx.com"
export TRAYLINX_AGENT_USER_ID="12345678-1234-1234-1234-123456789abc"from traylinx_auth_client import TraylinxAuthClient
client = TraylinxAuthClient(
client_id="your-client-id",
client_secret="your-client-secret",
api_base_url="https://auth.traylinx.com",
agent_user_id="12345678-1234-1234-1234-123456789abc",
timeout=30, # Request timeout in seconds
max_retries=3, # Maximum retry attempts
retry_delay=1.0, # Base delay between retries
cache_tokens=True, # Enable token caching
log_level="INFO" # Logging level
)from traylinx_auth_client import make_a2a_request
# GET request
data = make_a2a_request("GET", "https://other-agent.com/api/users")
# POST request with JSON data
result = make_a2a_request(
"POST",
"https://other-agent.com/api/process",
json={"items": ["item1", "item2"]},
timeout=60
)
# PUT request with custom headers
response = make_a2a_request(
"PUT",
"https://other-agent.com/api/update/123",
json={"status": "completed"},
headers={"X-Custom-Header": "value"}
)from traylinx_auth_client import get_agent_request_headers
import requests
# Get headers for calling other agents
headers = get_agent_request_headers()
# Make authenticated request
response = requests.get(
"https://other-agent.com/api/data",
headers=headers
)
# Headers include:
# {
# "X-Agent-Secret-Token": "your-agent-secret-token",
# "X-Agent-User-Id": "your-agent-user-id"
# }from fastapi import FastAPI, Request
from traylinx_auth_client import require_a2a_auth
app = FastAPI()
@app.get("/protected")
@require_a2a_auth
async def protected_endpoint(request: Request):
return {"message": "This endpoint requires A2A authentication"}
@app.post("/process")
@require_a2a_auth
async def process_data(request: Request, data: dict):
# This endpoint is automatically protected
return {"processed": data}from fastapi import FastAPI, Request, HTTPException
from traylinx_auth_client import validate_a2a_request
app = FastAPI()
@app.post("/manual-validation")
async def manual_validation(request: Request):
if not validate_a2a_request(request.headers):
raise HTTPException(status_code=401, detail="Invalid A2A token")
return {"message": "Token is valid"}from traylinx_auth_client import require_dual_auth, detect_auth_mode
@app.get("/flexible-auth")
@require_dual_auth # Supports both Bearer tokens and custom headers
async def flexible_endpoint(request: Request):
auth_mode = detect_auth_mode(request.headers)
return {"auth_mode": auth_mode, "message": "Authenticated successfully"}from traylinx_auth_client import TraylinxAuthClient
# Create client with custom configuration
client = TraylinxAuthClient(
timeout=60,
max_retries=5,
retry_delay=2.0
)
# Get individual tokens
access_token = client.get_access_token()
agent_secret_token = client.get_agent_secret_token()
# Get different header types
auth_headers = client.get_request_headers() # For auth service calls
agent_headers = client.get_agent_request_headers() # For agent calls
a2a_headers = client.get_a2a_headers() # A2A-compatible format
# Validate incoming tokens
is_valid = client.validate_token(
agent_secret_token="incoming-token",
agent_user_id="sender-agent-id"
)
# Context manager for automatic cleanup
with TraylinxAuthClient() as client:
response = client.rpc_health_check()from traylinx_auth_client import TraylinxAuthClient
client = TraylinxAuthClient()
# Built-in RPC methods
result = client.rpc_introspect_token(
agent_secret_token="token-to-validate",
agent_user_id="agent-id"
)
capabilities = client.rpc_get_capabilities()
health = client.rpc_health_check()
# Custom RPC calls
response = client.rpc_call(
method="custom_method",
params={"param1": "value1"},
rpc_url="https://custom-agent.com/a2a/rpc"
)
# RPC call with explicit credential control
response = client.rpc_call(
method="auth_service_method",
params={},
include_agent_credentials=False # Uses only access_token
)from traylinx_auth_client import (
TraylinxAuthClient,
AuthenticationError,
NetworkError,
ValidationError,
TokenExpiredError
)
try:
client = TraylinxAuthClient(
client_id="invalid-id",
client_secret="invalid-secret"
)
response = client.rpc_health_check()
except ValidationError as e:
print(f"Configuration error: {e}")
print(f"Error code: {e.error_code}")
except AuthenticationError as e:
print(f"Authentication failed: {e}")
print(f"Status code: {e.status_code}")
except NetworkError as e:
print(f"Network error: {e}")
if e.error_code == "TIMEOUT":
print("Request timed out - check network connectivity")
elif e.error_code == "RATE_LIMIT":
print("Rate limited - retry after delay")
except TokenExpiredError as e:
print(f"Token expired: {e}")
# Token will be automatically refreshed on next requestimport threading
from traylinx_auth_client import TraylinxAuthClient
# Thread-safe client usage
client = TraylinxAuthClient()
def worker_function(worker_id):
try:
# Each thread can safely use the same client
headers = client.get_agent_request_headers()
response = make_a2a_request("GET", f"https://api.com/data/{worker_id}")
print(f"Worker {worker_id}: {response}")
except Exception as e:
print(f"Worker {worker_id} error: {e}")
# Create multiple threads
threads = []
for i in range(10):
thread = threading.Thread(target=worker_function, args=(i,))
threads.append(thread)
thread.start()
# Wait for all threads to complete
for thread in threads:
thread.join()Make an authenticated A2A request to another agent.
Parameters:
method(str): HTTP method (GET, POST, PUT, DELETE, etc.)url(str): Target agent's URL**kwargs: Additional arguments forrequests.request()
Returns:
dict: JSON response from the target agent
Raises:
NetworkError: For network-related issuesAuthenticationError: For authentication failuresValidationError: For invalid parameters
Example:
response = make_a2a_request("POST", "https://agent.com/api", json={"key": "value"})Returns headers for calling the auth service (includes access_token).
Returns:
{
"Authorization": "Bearer <access_token>",
"X-Agent-Secret-Token": "<agent_secret_token>",
"X-Agent-User-Id": "<agent_user_id>"
}Returns headers for calling other agents (ONLY agent_secret_token).
Returns:
{
"X-Agent-Secret-Token": "<agent_secret_token>",
"X-Agent-User-Id": "<agent_user_id>"
}Returns A2A-compatible headers using Bearer token format.
Returns:
{
"Authorization": "Bearer <agent_secret_token>",
"X-Agent-User-Id": "<agent_user_id>"
}Decorator that protects FastAPI endpoints with A2A authentication.
Example:
@app.get("/protected")
@require_a2a_auth
async def protected_endpoint(request: Request):
return {"message": "Protected"}Enhanced decorator supporting both Bearer tokens and custom headers.
Validates incoming A2A request headers (custom format).
Parameters:
headers(dict): Request headers
Returns:
bool: True if valid, False otherwise
Validates incoming requests supporting both Bearer tokens and custom headers.
Detect authentication mode from request headers.
Returns:
str: 'bearer', 'custom', or 'none'
TraylinxAuthClient(
client_id: str = None, # OAuth client ID
client_secret: str = None, # OAuth client secret
api_base_url: str = None, # Traylinx API base URL
agent_user_id: str = None, # Agent user UUID
timeout: int = 30, # Request timeout (seconds)
max_retries: int = 3, # Maximum retry attempts
retry_delay: float = 1.0, # Base retry delay (seconds)
cache_tokens: bool = True, # Enable token caching
log_level: str = "INFO" # Logging level
)Parameters:
- All parameters default to corresponding environment variables
timeout: Request timeout in seconds (default: 30)max_retries: Maximum retry attempts for failed requests (default: 3)retry_delay: Base delay between retries in seconds (default: 1.0)cache_tokens: Whether to cache tokens in memory (default: True)log_level: Logging level - DEBUG, INFO, WARN, ERROR (default: "INFO")
Raises:
ValidationError: If configuration parameters are invalid
Get current access token for calling auth service.
Returns:
str: Valid access token
Raises:
TokenExpiredError: If token is unavailableAuthenticationError: If token fetch fails
Get current agent secret token for agent-to-agent communication.
Returns:
str: Valid agent secret token
Raises:
TokenExpiredError: If token is unavailableAuthenticationError: If token fetch fails
Get headers for calling the auth service (includes access_token).
Get headers for calling other agents (ONLY agent_secret_token).
Get A2A-compatible authentication headers using Bearer token format.
Validate an agent secret token against the auth service.
Parameters:
agent_secret_token(str): Token to validateagent_user_id(str): Agent user ID associated with token
Returns:
bool: True if token is valid and active
Raises:
AuthenticationError: If validation request failsNetworkError: For network-related issues
Validate A2A request supporting both Bearer tokens and custom headers.
Detect authentication mode from request headers.
rpc_call(method: str, params: dict, rpc_url: str = None, include_agent_credentials: bool = None) -> dict
Make a JSON-RPC call with automatic credential detection.
Parameters:
method(str): RPC method nameparams(dict): RPC method parametersrpc_url(str, optional): Custom RPC endpoint URLinclude_agent_credentials(bool, optional): Whether to include agent credentials
Returns:
dict: JSON-RPC response
Raises:
ValidationError: For invalid RPC requests or parametersAuthenticationError: For authentication failuresNetworkError: For network issuesTraylinxAuthError: For RPC-specific errors
Introspect a token via JSON-RPC.
Get agent capabilities via JSON-RPC.
Perform health check via JSON-RPC.
Close the HTTP session and clean up resources.
Support for context manager usage:
with TraylinxAuthClient() as client:
response = client.rpc_health_check()
# Automatically cleaned upBase exception class for all TraylinxAuthClient errors.
Attributes:
error_code(str): Specific error codestatus_code(int): HTTP status code (if applicable)
Raised for input validation failures.
Raised for authentication-related failures.
Raised when tokens are expired or unavailable.
Raised for network-related issues (timeouts, connection errors, etc.).
Pydantic model for configuration validation.
Fields:
client_id(str): OAuth client IDclient_secret(str): OAuth client secretapi_base_url(HttpUrl): Traylinx API base URLagent_user_id(str): Agent user UUIDtimeout(int): Request timeout in secondsmax_retries(int): Maximum retry attemptsretry_delay(float): Base retry delaycache_tokens(bool): Enable token cachinglog_level(str): Logging level
Traylinx uses a dual-token authentication system for enhanced security:
access_token: Used for calling Traylinx auth service endpointsagent_secret_token: Used for agent-to-agent communication
sequenceDiagram
participant Client as TraylinxAuthClient
participant Auth as Traylinx Sentinel API
participant Agent as Target Agent
Note over Client,Auth: Initial Authentication
Client->>Auth: POST /oauth/token (client_credentials)
Auth-->>Client: {access_token, agent_secret_token, expires_in}
Note over Client,Auth: Calling Auth Service
Client->>Auth: GET /a2a/rpc (Authorization: Bearer access_token)
Auth-->>Client: Response
Note over Client,Agent: Calling Other Agents
Client->>Agent: POST /endpoint (X-Agent-Secret-Token: agent_secret_token)
Agent->>Auth: Validate token
Auth-->>Agent: Token valid
Agent-->>Client: Response
| Scenario | access_token | agent_secret_token | Headers |
|---|---|---|---|
| Auth service calls | β Required | β Not used | Authorization: Bearer <access_token> |
| Agent-to-agent calls | β Not used | β Required | X-Agent-Secret-Token: <agent_secret_token> |
| A2A compatible calls | β Not used | β Required | Authorization: Bearer <agent_secret_token> |
| Token validation | β Required | β Not used | Authorization: Bearer <access_token> |
- Token Acquisition: Automatically fetches tokens using OAuth2
client_credentialsgrant - Token Caching: Tokens cached in memory with thread-safe access
- Automatic Refresh: Expired tokens refreshed automatically before requests
- Error Recovery: Handles token expiration and authentication failures gracefully
Example token response:
{
"access_token": "eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...",
"agent_secret_token": "TqlJuJi5aJa7lz8rg9zWjbRDChND8m9PMr4bsn...",
"token_type": "Bearer",
"expires_in": 7200,
"scope": "a2a"
}The client provides robust error handling with specific exception types:
from traylinx_auth_client import (
TraylinxAuthError, # Base exception
ValidationError, # Configuration/input validation
AuthenticationError, # Authentication failures
TokenExpiredError, # Token expiration
NetworkError # Network/connectivity issues
)- Exponential Backoff: Automatic retries with increasing delays
- Configurable Retries: Set
max_retriesandretry_delayparameters - Smart Retry Logic: Only retries on transient failures (429, 5xx errors)
- Connection Pooling: Efficient connection reuse for better performance
| Error Type | HTTP Status | Retry | Description |
|---|---|---|---|
ValidationError |
400 | β | Invalid configuration or parameters |
AuthenticationError |
401 | β | Invalid credentials or expired tokens |
NetworkError (Rate Limit) |
429 | β | Rate limiting - automatic retry with backoff |
NetworkError (Timeout) |
408 | β | Request timeout - configurable retry |
NetworkError (Server Error) |
5xx | β | Server errors - automatic retry |
NetworkError (Connection) |
0 | β | Connection failures - automatic retry |
import logging
from traylinx_auth_client import TraylinxAuthClient, NetworkError
# Configure logging
logging.basicConfig(level=logging.INFO)
client = TraylinxAuthClient(log_level="INFO")
try:
response = client.rpc_health_check()
except NetworkError as e:
if e.error_code == "RATE_LIMIT":
logging.warning(f"Rate limited: {e}")
# Implement backoff strategy
elif e.error_code == "TIMEOUT":
logging.error(f"Request timeout: {e}")
# Check network connectivity
else:
logging.error(f"Network error: {e}")β DO:
- Store credentials in environment variables or secure vaults
- Use different credentials for different environments (dev/staging/prod)
- Regularly rotate client credentials
- Monitor authentication failures and unusual patterns
- Use HTTPS for all communications
β DON'T:
- Hard-code credentials in source code
- Log sensitive data (tokens, secrets, passwords)
- Share credentials between different applications
- Use production credentials in development/testing
# β
Good: Use environment variables
client = TraylinxAuthClient() # Reads from env vars
# β
Good: Use secure credential management
import os
from your_vault import get_secret
client = TraylinxAuthClient(
client_id=os.getenv("TRAYLINX_CLIENT_ID"),
client_secret=get_secret("traylinx_client_secret"),
api_base_url=os.getenv("TRAYLINX_API_BASE_URL"),
agent_user_id=os.getenv("TRAYLINX_AGENT_USER_ID")
)
# β Bad: Hard-coded credentials
client = TraylinxAuthClient(
client_id="hardcoded-id", # Never do this!
client_secret="hardcoded-secret" # Never do this!
)- TLS/HTTPS: All communications use HTTPS with certificate validation
- Request Signing: Tokens provide request authenticity
- Timeout Protection: Configurable timeouts prevent hanging requests
- Rate Limiting: Built-in protection against rate limiting
import logging
# β
Safe logging - no sensitive data
logging.info("Authentication successful for agent %s", agent_user_id)
logging.error("Authentication failed with status %d", response.status_code)
# β Unsafe logging - exposes sensitive data
logging.info("Token: %s", access_token) # Never log tokens!
logging.debug("Secret: %s", client_secret) # Never log secrets!Problem: AuthenticationError: Invalid credentials
Solutions:
- Verify environment variables are set correctly
- Check client_id and client_secret are valid
- Ensure API base URL is correct
- Verify agent_user_id is a valid UUID
# Check environment variables
echo $TRAYLINX_CLIENT_ID
echo $TRAYLINX_CLIENT_SECRET
echo $TRAYLINX_API_BASE_URL
echo $TRAYLINX_AGENT_USER_IDProblem: NetworkError: Connection failed
Solutions:
- Check network connectivity to API endpoint
- Verify firewall/proxy settings
- Increase timeout value
- Check DNS resolution
# Test connectivity
import requests
response = requests.get("https://your-api-base-url/health")
print(response.status_code)Problem: ValidationError: Configuration validation failed
Solutions:
- Ensure agent_user_id is a valid UUID format
- Verify API base URL is a valid HTTPS URL
- Check all required parameters are provided
# Validate UUID format
import uuid
try:
uuid.UUID("your-agent-user-id")
print("Valid UUID")
except ValueError:
print("Invalid UUID format")Problem: TokenExpiredError: Token is not available
Solutions:
- Check if initial authentication succeeded
- Verify network connectivity during token refresh
- Check if credentials are still valid
import logging
# Enable debug logging
logging.basicConfig(level=logging.DEBUG)
client = TraylinxAuthClient(log_level="DEBUG")
# This will show detailed request/response information# Optimize for high-throughput scenarios
client = TraylinxAuthClient(
timeout=60, # Longer timeout for slow networks
max_retries=5, # More retries for reliability
retry_delay=0.5, # Faster initial retry
cache_tokens=True # Enable token caching
)
# Use context manager for automatic cleanup
with TraylinxAuthClient() as client:
# Client automatically cleaned up after use
response = client.rpc_health_check()A: The access_token is used for calling Traylinx auth service endpoints, while agent_secret_token is used for agent-to-agent communication. They serve different purposes in the dual-token authentication system.
A: Token expiration is handled automatically. The client will refresh tokens before they expire and retry failed requests with fresh tokens.
A: Yes! The client is thread-safe and can be safely used across multiple threads. Token management is protected with locks.
A: Use the max_retries and retry_delay parameters:
client = TraylinxAuthClient(
max_retries=5, # Retry up to 5 times
retry_delay=2.0 # Start with 2-second delay
)A: The client will retry requests with exponential backoff. If all retries fail, it will raise a NetworkError with details about the failure.
A: Use the @require_a2a_auth decorator or validate_a2a_request() function:
@app.post("/endpoint")
@require_a2a_auth
async def my_endpoint(request: Request):
return {"status": "authenticated"}# Clone the repository
git clone https://github.com/traylinx/traylinx-auth-client-py.git
cd traylinx-auth-client-py
# Install Poetry (if not already installed)
curl -sSL https://install.python-poetry.org | python3 -
# Install dependencies
poetry install
# Activate virtual environment
poetry shell# Run all tests
poetry run pytest
# Run tests with coverage report
poetry run pytest --cov=traylinx_auth_client --cov-report=html
# Run specific test file
poetry run pytest tests/test_client.py
# Run tests with verbose output
poetry run pytest -v
# Run tests and generate coverage report
poetry run pytest --cov=traylinx_auth_client --cov-report=term-missing# Format code with Black
poetry run black traylinx_auth_client/ tests/
# Lint with flake8
poetry run flake8 traylinx_auth_client/ tests/
# Type checking with mypy
poetry run mypy traylinx_auth_client/
# Security scan with bandit
poetry run bandit -r traylinx_auth_client/# Build package
poetry build
# Check package contents
tar -tzf dist/traylinx-auth-client-*.tar.gz
# Test publish to TestPyPI
poetry config repositories.testpypi https://test.pypi.org/legacy/
poetry publish -r testpypi
# Publish to PyPI
poetry publishtraylinx_auth_client_py/
βββ traylinx_auth_client/
β βββ __init__.py # Public API exports
β βββ main.py # High-level functions and decorators
β βββ client.py # Core TraylinxAuthClient class
β βββ config.py # Configuration validation with Pydantic
β βββ exceptions.py # Custom exception hierarchy
βββ tests/
β βββ test_client.py # Client class tests
β βββ test_config.py # Configuration validation tests
β βββ test_exceptions.py # Exception handling tests
β βββ test_main.py # High-level function tests
βββ pyproject.toml # Project configuration and dependencies
βββ README.md # This file
βββ CHANGELOG.md # Version history
βββ CONTRIBUTING.md # Contribution guidelines
βββ SECURITY.md # Security policy
βββ LICENSE # MIT license
- Create Feature Branch:
git checkout -b feature/new-feature - Write Tests First: Add tests in appropriate test file
- Implement Feature: Add implementation with proper docstrings
- Update Documentation: Update README and docstrings
- Run Quality Checks: Ensure tests pass and code quality is maintained
- Submit Pull Request: Include description of changes and test results
- Unit Tests: Test individual functions and methods
- Integration Tests: Test end-to-end authentication flows
- Error Tests: Test all error conditions and edge cases
- Mock External Services: Use mocks for HTTP requests to auth service
- Thread Safety Tests: Test concurrent access patterns
We welcome contributions! Please see CONTRIBUTING.md for detailed guidelines.
- Fork the repository on GitHub
- Clone your fork locally
- Create a feature branch from
main - Make your changes with tests
- Run the test suite to ensure everything works
- Submit a pull request with a clear description
- π Bug Fixes: Fix issues and improve reliability
- β¨ New Features: Add new functionality
- π Documentation: Improve docs and examples
- π§ Performance: Optimize performance and efficiency
- π§ͺ Tests: Add or improve test coverage
- π¨ Code Quality: Refactoring and code improvements
- Follow PEP 8 style guidelines
- Add type hints for all public functions
- Write comprehensive docstrings
- Maintain test coverage above 90%
- Use meaningful variable and function names
This project is licensed under the MIT License - see the LICENSE file for details.
- β Commercial use allowed
- β Modification allowed
- β Distribution allowed
- β Private use allowed
- β No warranty provided
- β No liability accepted
- π Documentation: Traylinx Developer Docs
- π Bug Reports: GitHub Issues
- π¬ Discussions: GitHub Discussions
- π§ Email Support: [email protected]
When reporting issues, please include:
- Python version and operating system
- Library version (
pip show traylinx-auth-client) - Minimal code example that reproduces the issue
- Full error traceback if applicable
- Expected vs actual behavior
We welcome feature requests! Please:
- Check existing issues to avoid duplicates
- Describe the use case and problem you're solving
- Provide examples of how the feature would be used
- Consider contributing the implementation
For security-related issues, please follow our Security Policy:
- π Private Disclosure: Email [email protected]
- β±οΈ Response Time: We aim to respond within 24 hours
- π‘οΈ Responsible Disclosure: We follow coordinated disclosure practices
Made with β€οΈ by the Traylinx Team