mcpField
P
pydantic-ai-dependency-injection

pydantic-ai-dependency-injection

github.com/existential-birdsエージェントワークフロー85140 回インストール11 週間前に更新
リポジトリ

pydantic-ai-dependency-injection について

PydanticAI 開発者向けの依存性注入ガイドです。RunContext を使って接続やクライアントをツールや指示に渡す方法を説明し、型注釈、実行時引数、テスト用オブジェクト、依存定義の確認を扱います。

主な機能

  1. 01RunContext による依存関係の受け渡し
  2. 02dataclass または Pydantic モデルでの定義
  3. 03Agent へのジェネリック型注釈
  4. 04モック依存関係を使ったテスト
  5. 05依存フィールドと実行時引数の確認

活用シーン

  1. 01ツールにデータベース接続を渡す
  2. 02指示にユーザーコンテキストを注入
  3. 03Agent の依存型を確認
  4. 04モックで依存ロジックをテスト

インストール

コマンドをコピーしてターミナルで実行します。

$npx skills add https://github.com/existential-birds/beagle --skill pydantic-ai-dependency-injection

インストール後、Claude Code で/pydantic-ai-dependency-injectionを入力して読み込まれたか確認します。

詳細

PydanticAI の依存性注入

基本パターン

依存関係は RunContext を通じて渡されます。

from dataclasses import dataclass
from pydantic_ai import Agent, RunContext

@dataclass
class Deps:
    db: DatabaseConn
    api_client: HttpClient
    user_id: int

agent = Agent(
    'openai:gpt-4o',
    deps_type=Deps,  # Type for static analysis
)

@agent.tool
async def get_user_balance(ctx: RunContext[Deps]) -> float:
    """Get the current user's account balance."""
    return await ctx.deps.db.get_balance(ctx.deps.user_id)

# At runtime, provide deps
result = await agent.run(
    'What is my balance?',
    deps=Deps(db=db_conn, api_client=client, user_id=123)
)

依存関係の定義

データクラスまたは Pydantic モデルを使用します。

from dataclasses import dataclass
from pydantic import BaseModel

# Dataclass (recommended for simplicity)
@dataclass
class Deps:
    db: DatabaseConnection
    cache: CacheClient
    user_context: UserContext

# Pydantic model (if you need validation)
class Deps(BaseModel):
    api_key: str
    endpoint: str
    timeout: int = 30

依存関係へのアクセス

ツールや指示内でアクセスします。

@agent.tool
async def query_database(ctx: RunContext[Deps], query: str) -> list[dict]:
    """Run a database query."""
    return await ctx.deps.db.execute(query)

@agent.instructions
async def add_user_context(ctx: RunContext[Deps]) -> str:
    user = await ctx.deps.db.get_user(ctx.deps.user_id)
    return f"User name: {user.name}, Role: {user.role}"

@agent.system_prompt
def add_permissions(ctx: RunContext[Deps]) -> str:
    return f"User has permissions: {ctx.deps.permissions}"

型安全性

ジェネリクスを使った完全な型チェックが可能です。

# Explicit agent type annotation
agent: Agent[Deps, OutputModel] = Agent(
    'openai:gpt-4o',
    deps_type=Deps,
    output_type=OutputModel,
)

# Now these are type-checked:
# - ctx.deps in tools is typed as Deps
# - result.output is typed as OutputModel
# - agent.run() requires deps: Deps

依存関係がない場合のパターン

依存関係が不要な場合は、次のようにします。

# Option 1: No deps_type (defaults to NoneType)
agent = Agent('openai:gpt-4o')
result = agent.run_sync('Hello')  # No deps needed

# Option 2: Explicit None for type checker
agent: Agent[None, str] = Agent('openai:gpt-4o')
result = agent.run_sync('Hello', deps=None)

# In tool_plain, no context access
@agent.tool_plain
def simple_calc(a: int, b: int) -> int:
    return a + b

完全な例

from dataclasses import dataclass
from httpx import AsyncClient
from pydantic import BaseModel
from pydantic_ai import Agent, RunContext

@dataclass
class WeatherDeps:
    client: AsyncClient
    api_key: str

class WeatherReport(BaseModel):
    location: str
    temperature: float
    conditions: str

agent: Agent[WeatherDeps, WeatherReport] = Agent(
    'openai:gpt-4o',
    deps_type=WeatherDeps,
    output_type=WeatherReport,
    instructions='You are a weather assistant.',
)

@agent.tool
async def get_weather(
    ctx: RunContext[WeatherDeps],
    city: str
) -> dict:
    """Fetch weather data for a city."""
    response = await ctx.deps.client.get(
        f'https://api.weather.com/{city}',
        headers={'Authorization': ctx.deps.api_key}
    )
    return response.json()

async def main():
    async with AsyncClient() as client:
        deps = WeatherDeps(client=client, api_key='secret')
        result = await agent.run('Weather in London?', deps=deps)
        print(result.output.temperature)

テスト用のオーバーライド

from pydantic_ai.models.test import TestModel

# Create mock dependencies
mock_deps = Deps(
    db=MockDatabase(),
    api_client=MockClient(),
    user_id=999
)

# Override model and deps for testing
with agent.override(model=TestModel(), deps=mock_deps):
    result = agent.run_sync('Test prompt')

ゲート

エージェントが正しく動作すると判断する前に、以下を順番に実行してください。各ステップには客観的な合格条件があります。

  1. すべてのアクセスに依存関係を用意する — ツール、@agent.instructions、@agent.system_prompt から、すべての ctx.deps.<attr>(およびネストされた使用箇所)を収集します。合格条件: 各 <attr> が deps_type に存在すること(また、mypy/pyright を使っている場合は、Agent[DepsType, …] の静的チェックに合格すること)。
  2. 依存関係が必要なすべての実行に依存関係を渡す — 合格条件: それらのツールを実行する各 agent.run / run_sync 経路で、型が deps_type と一致する deps= を渡していること(エージェントに依存関係が本当にない場合を除き、None は不可)。
  3. テストで依存関係の構造を固定する — 合格条件: agent.override を使用するテストでは、本番環境の Deps と同じフィールドと型を持つ deps= 値を渡していること(テスト対象のツールが不足しているフィールドにアクセスしない場合を除き、部分的なモックは不可)。

ベストプラクティス

  1. 依存関係を不変に保つ:凍結データクラスまたは Pydantic モデルを使用します
  2. 認証情報ではなく接続を渡す:依存関係には初期化済みのクライアントを保持させます
  3. エージェントに型を指定する:完全な型安全性を確保するには Agent[DepsType, OutputType] を使用します
  4. 依存関係を適切なスコープで管理する:リクエストの開始時に依存関係を作成し、終了後に閉じます

コメント