gunn.adapters.web package

Submodules

gunn.adapters.web.cli module

CLI for running the Web adapter server.

This module provides a command-line interface for starting the FastAPI-based Web adapter server with configurable options.

class gunn.adapters.web.cli.AuthTokenConfig(**data)[ソース]

ベースクラス: BaseModel

Configuration for authentication tokens.

パラメータ:
token: str
world_id: str
agent_id: str
permissions: list[str]
expires_at: float | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gunn.adapters.web.cli.WebServerConfig(**data)[ソース]

ベースクラス: BaseModel

Configuration for the web server.

パラメータ:
host: str
port: int
workers: int
log_level: str
rate_limit_requests: int
rate_limit_window: int
auth_tokens: list[AuthTokenConfig]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

gunn.adapters.web.health module

Health check endpoints for web adapter.

gunn.adapters.web.health.setup_health_checker(config)[ソース]

Setup global health checker instance.

戻り値の型:

HealthChecker

パラメータ:

config (HealthCheckConfig)

gunn.adapters.web.health.get_health_checker()[ソース]

Get the global health checker instance.

戻り値の型:

HealthChecker

gunn.adapters.web.health.create_health_router()[ソース]

Create health check router.

戻り値の型:

APIRouter

gunn.adapters.web.server module

FastAPI-based Web adapter for external system integration.

This module provides REST and WebSocket endpoints for external systems to interact with the multi-agent simulation core, including authentication, authorization, and rate limiting.

Requirements addressed: - 8.3: Expose REST/WebSocket endpoints for external system integration - 12.1: Authenticate using token-based auth and authorize per agent_id/world_id - 12.4: Enforce rate limiting per tenant and agent

class gunn.adapters.web.server.AuthToken(**data)[ソース]

ベースクラス: BaseModel

Authentication token with permissions.

パラメータ:
token: str
world_id: str
agent_id: str
permissions: list[str]
expires_at: float | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gunn.adapters.web.server.IntentRequest(**data)[ソース]

ベースクラス: BaseModel

Request model for intent submission.

パラメータ:
kind: str
payload: dict[str, Any]
priority: int
context_seq: int
schema_version: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gunn.adapters.web.server.IntentResponse(**data)[ソース]

ベースクラス: BaseModel

Response model for intent submission.

パラメータ:

  • req_id (str)

  • status (str)

  • message (str | None)

req_id: str
status: str
message: str | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gunn.adapters.web.server.ObservationResponse(**data)[ソース]

ベースクラス: BaseModel

Response model for observation requests.

パラメータ:
view_seq: int
patches: list[dict[str, Any]]
context_digest: str
schema_version: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gunn.adapters.web.server.ErrorResponse(**data)[ソース]

ベースクラス: BaseModel

Error response model.

パラメータ:
error: str
message: str
details: dict[str, Any] | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class gunn.adapters.web.server.RateLimiter(max_requests=100, window_seconds=60)[ソース]

ベースクラス: object

Simple in-memory rate limiter.

パラメータ:

  • max_requests (int)

  • window_seconds (int)

is_allowed(key)[ソース]

Check if request is allowed for the given key.

戻り値の型:

bool

パラメータ:

key (str)

Args:

key: Rate limiting key (e.g., "world_id:agent_id")

Returns:

True if request is allowed

class gunn.adapters.web.server.WebAdapter(orchestrator, auth_tokens=None, rate_limit_requests=100, rate_limit_window=60)[ソース]

ベースクラス: object

FastAPI-based Web adapter for external integration.

パラメータ:
add_auth_token(token, world_id, agent_id, permissions, expires_at=None)[ソース]

Add authentication token.

戻り値の型:

None

パラメータ:
Args:

token: Authentication token world_id: World ID this token is valid for agent_id: Agent ID this token is valid for permissions: List of permissions expires_at: Optional expiration timestamp

remove_auth_token(token)[ソース]

Remove authentication token.

戻り値の型:

None

パラメータ:

token (str)

Args:

token: Authentication token to remove

async shutdown()[ソース]

Shutdown the web adapter.

戻り値の型:

None

gunn.adapters.web.server.create_web_adapter(orchestrator, auth_tokens=None, rate_limit_requests=100, rate_limit_window=60)[ソース]

Create a Web adapter instance.

戻り値の型:

WebAdapter

パラメータ:
Args:

orchestrator: Orchestrator instance auth_tokens: Authentication tokens mapping rate_limit_requests: Rate limit requests per window rate_limit_window: Rate limit window in seconds

Returns:

WebAdapter instance

Module contents

Web adapter for REST and WebSocket endpoints.

This module provides FastAPI-based REST and WebSocket endpoints for external system integration with authentication, authorization, and rate limiting.

class gunn.adapters.web.AuthToken(**data)[ソース]

ベースクラス: BaseModel

Authentication token with permissions.

パラメータ:
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

token: str
world_id: str
agent_id: str
permissions: list[str]
expires_at: float | None
class gunn.adapters.web.ErrorResponse(**data)[ソース]

ベースクラス: BaseModel

Error response model.

パラメータ:
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

error: str
message: str
details: dict[str, Any] | None
class gunn.adapters.web.IntentRequest(**data)[ソース]

ベースクラス: BaseModel

Request model for intent submission.

パラメータ:
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

kind: str
payload: dict[str, Any]
priority: int
context_seq: int
schema_version: str
class gunn.adapters.web.IntentResponse(**data)[ソース]

ベースクラス: BaseModel

Response model for intent submission.

パラメータ:

  • req_id (str)

  • status (str)

  • message (str | None)

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

req_id: str
status: str
message: str | None
class gunn.adapters.web.ObservationResponse(**data)[ソース]

ベースクラス: BaseModel

Response model for observation requests.

パラメータ:
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

view_seq: int
patches: list[dict[str, Any]]
context_digest: str
schema_version: str
class gunn.adapters.web.WebAdapter(orchestrator, auth_tokens=None, rate_limit_requests=100, rate_limit_window=60)[ソース]

ベースクラス: object

FastAPI-based Web adapter for external integration.

パラメータ:
add_auth_token(token, world_id, agent_id, permissions, expires_at=None)[ソース]

Add authentication token.

戻り値の型:

None

パラメータ:
Args:

token: Authentication token world_id: World ID this token is valid for agent_id: Agent ID this token is valid for permissions: List of permissions expires_at: Optional expiration timestamp

remove_auth_token(token)[ソース]

Remove authentication token.

戻り値の型:

None

パラメータ:

token (str)

Args:

token: Authentication token to remove

async shutdown()[ソース]

Shutdown the web adapter.

戻り値の型:

None

gunn.adapters.web.create_web_adapter(orchestrator, auth_tokens=None, rate_limit_requests=100, rate_limit_window=60)[ソース]

Create a Web adapter instance.

戻り値の型:

WebAdapter

パラメータ:
Args:

orchestrator: Orchestrator instance auth_tokens: Authentication tokens mapping rate_limit_requests: Rate limit requests per window rate_limit_window: Rate limit window in seconds

Returns:

WebAdapter instance