### Multi-component Logging Example

Source: https://jolovicdev.github.io/blackgeorge/logging/index

An example illustrating how to create and use component-specific loggers with shared base context.

```APIDOC
### Multi-component logging

```python
from blackgeorge.logging import get_logger

# Create base logger for the application
base_logger = get_logger("myapp").with_context(
    app="myapp",
    version="1.0.0"
)

# Create component-specific loggers
db_logger = base_logger.with_context(component="database")
api_logger = base_logger.with_context(component="api")
worker_logger = base_logger.with_context(component="worker")

# All logs include app and version context
db_logger.info("Connected to database")
api_logger.info("Received request")
worker_logger.info("Processing job")
```
```

--------------------------------

### Multi-agent Analysis Setup with Collaboration

Source: https://context7_llms

Demonstrates the setup for a multi-agent analysis scenario using Blackgeorge. It involves creating Channel and Blackboard instances, defining necessary tools, and initializing workers and a workforce for collaborative execution.

```python
from blackgeorge import Desk, Job, Worker, Workforce
from blackgeorge.collaboration import Channel, Blackboard
from blackgeorge.collaboration.tools import (
    channel_send_tool,
    channel_receive_tool,
    blackboard_read_tool,
    blackboard_write_tool,
)

# Create collaboration primitives
channel = Channel()
blackboard = Blackboard()

```

--------------------------------

### Tool Execution Tracking Example

Source: https://jolovicdev.github.io/blackgeorge/logging/index

An example showing how to track tool execution, including logging queries and results.

```APIDOC
### Tool execution tracking

```python
from blackgeorge.tools import tool
from blackgeorge.logging import get_logger

logger = get_logger("tools").with_context(component="database")

@tool()
def query_database(sql: str) -> list[dict]:
    logger.info("Executing query", sql=sql)
    try:
        results = execute_sql(sql)
        logger.info("Query completed", row_count=len(results))
        return results
    except Exception as e:
        logger.error("Query failed", sql=sql, error=str(e))
        raise
```
```

--------------------------------

### Serve Documentation Site Locally

Source: https://context7_llms

Starts a local development server for the MkDocs documentation site. This allows for live previewing of documentation changes.

```text
uv run mkdocs serve
```

--------------------------------

### Request Logging Example

Source: https://jolovicdev.github.io/blackgeorge/logging/index

An example demonstrating how to log request details throughout its lifecycle.

```APIDOC
### Request logging

```python
from blackgeorge.logging import get_logger

logger = get_logger("api").with_context(service="user-service")

def handle_request(request):
    request_logger = logger.with_context(
        request_id=request.id,
        user_id=request.user_id
    )

    request_logger.info("Request started", path=request.path)
    try:
        response = process_request(request)
        request_logger.info("Request completed", status=response.status_code)
        return response
    except Exception as e:
        request_logger.error("Request failed", error=str(e))
        raise
```
```

--------------------------------

### Install Development Dependencies

Source: https://context7_llms

Installs the development dependencies for the project using uv pip. The `.[dev]` flag ensures that all optional development-specific packages are included.

```text
uv pip install -e .[dev]
```

--------------------------------

### Shell Command to Run Python Example

Source: https://context7_llms

Executes the main Python script for the coding agent example. This command assumes the script is located at `examples/coding_agent/run.py` and that the necessary environment variables (like API keys) are already set.

```bash
python examples/coding_agent/run.py
```

--------------------------------

### Install Blackgeorge using pip

Source: https://context7_llms

Installs the Blackgeorge package using pip. This is the standard Python package installer.

```text
pip install blackgeorge
```

--------------------------------

### Install Blackgeorge using uv

Source: https://context7_llms

Installs the Blackgeorge package using the 'uv' package manager. This is a quick way to add the library to your project.

```text
uv add blackgeorge
```

--------------------------------

### Creating a Blackboard in Python

Source: https://context7_llms

Demonstrates the basic setup for using the Blackboard, a collaboration primitive for sharing state across multiple workers. This code snippet shows how to instantiate a Blackboard object.

```python
from blackgeorge.collaboration import Blackboard

blackboard = Blackboard()
```

--------------------------------

### Log Request Lifecycle with Context

Source: https://jolovicdev.github.io/blackgeorge/logging/index

An example illustrating how to log the lifecycle of a web request, including start and completion messages, with dynamic context like request ID and user ID. It also handles errors during request processing.

```python
from blackgeorge.logging import get_logger

logger = get_logger("api").with_context(service="user-service")

def handle_request(request):
    request_logger = logger.with_context(
        request_id=request.id,
        user_id=request.user_id
    )

    request_logger.info("Request started", path=request.path)
    try:
        response = process_request(request)
        request_logger.info("Request completed", status=response.status_code)
        return response
    except Exception as e:
        request_logger.error("Request failed", error=str(e))
        raise
```

--------------------------------

### Run a Basic Worker with a Desk

Source: https://context7_llms

Demonstrates the basic usage of Blackgeorge by creating a Desk, a Worker, and a Job. The Desk then runs the Worker on the Job, and the report's content is printed. Requires the 'blackgeorge' library to be installed.

```python
from blackgeorge import Desk, Job, Worker

desk = Desk(model="openai/gpt-5-nano")
worker = Worker(name="Researcher")
job = Job(input="Summarize this topic", expected_output="A short summary")

report = desk.run(worker, job)
print(report.content)
```

--------------------------------

### Logging within Blackgeorge Tools

Source: https://jolovicdev.github.io/blackgeorge/logging/index

Shows how to integrate Blackgeorge's structured logger within a tool function. The example logs the start and success or failure of a file reading operation, including relevant context like file path and error messages.

```python
from blackgeorge.tools import tool
from blackgeorge.logging import get_logger

logger = get_logger("tools").with_context(component="file_operations")

@tool()
def read_file(file_path: str) -> str:
    logger.info("Reading file", file_path=file_path)
    try:
        with open(file_path) as f:
            content = f.read()
        logger.info("File read successfully", file_path=file_path, size=len(content))
        return content
    except Exception as e:
        logger.error("Failed to read file", file_path=file_path, error=str(e))
        raise
```

--------------------------------

### Workforce Initialization with Collaboration Primitives

Source: https://context7_llms

Demonstrates how to initialize a Blackgeorge Workforce with Channel and Blackboard instances. This setup enables workers within the workforce to utilize the collaboration features for communication and shared state.

```python
from blackgeorge import Worker, Workforce
from blackgeorge.collaboration import Channel, Blackboard

channel = Channel()
blackboard = Blackboard()

worker_a = Worker(name="worker_a")
worker_b = Worker(name="worker_b")

workforce = Workforce(
    [worker_a, worker_b],
    mode="collaborate",
    channel=channel,
    blackboard=blackboard,
)
```

--------------------------------

### Conditional Pre-Hooks in Python

Source: https://context7_llms

Demonstrates how to use a pre-hook to conditionally execute logic before a tool is called. This example logs a message only when an 'admin_' prefixed tool is invoked, checking the caller's metadata.

```python
from blackgeorge.tools import tool

def conditional_pre(call):
    # Only log for specific tools
    if call.name.startswith("admin_"):
        print(f"Admin tool {call.name} called by {call.metadata.get('user')}")

@tool(pre=(conditional_pre,))
def admin_delete_user(user_id: str) -> str:
    return f"Deleted {user_id}"
```

--------------------------------

### Integration with Datadog

Source: https://jolovicdev.github.io/blackgeorge/logging/index

Example of logging metrics compatible with Datadog.

```APIDOC
### Datadog

```python
logger.info("Metric recorded", metric_name="request.duration", value=123.45, tags=["api:v1"])
```
```

--------------------------------

### Preserve Example Changes in Blackgeorge Coding Agent

Source: https://context7_llms

The PRESERVE_EXAMPLE_CHANGES environment variable controls whether file changes made by the coding agent example are preserved. Setting it to '1' preserves changes; otherwise, changes are restored to their original state.

```bash
# Preserve example changes (default: changes are restored)
export PRESERVE_EXAMPLE_CHANGES="1"
```

--------------------------------

### Implement Timing Pre-hook for Tools in Python

Source: https://context7_llms

Demonstrates a pre-hook in Python designed for timing tool execution. The `start_timer` function records the start time associated with a unique tool call ID before the tool runs. This data can be used later to calculate execution duration.

```python
import time
from blackgeorge.tools import tool

timing_data = {}

def start_timer(call):
    timing_data[call.id] = time.time()

@tool(pre=(start_timer,))
def expensive_operation(n: int) -> int:
    return sum(range(n))
```

--------------------------------

### Implement Custom BaseModelAdapter in Python

Source: https://context7_llms

Demonstrates how to implement the `BaseModelAdapter` interface in Python to create a custom adapter for Blackgeorge. This involves defining synchronous (`complete`) and asynchronous (`acomplete`) methods for model interaction. The example shows instantiating the `Desk` class with the custom adapter.

```python
from blackgeorge.adapters.base import BaseModelAdapter

class CustomAdapter(BaseModelAdapter):
    def complete(
        self,
        model: str,
        messages: list[dict],
        tools: list[dict] | None,
        tool_choice: str | None,
        temperature: float | None,
        max_tokens: int | None,
        stream: bool,
        stream_options: dict | None,
    ) -> ModelResponse:
        # Implementation
        pass

    async def acomplete(
        self,
        model: str,
        messages: list[dict],
        tools: list[dict] | None,
        tool_choice: str | None,
        temperature: float | None,
        max_tokens: int | None,
        stream: bool,
        stream_options: dict | None,
    ) -> ModelResponse:
        # Implementation
        pass

desk = Desk(adapter=CustomAdapter())
```

--------------------------------

### Connect to Local MCP Server via stdio in Python

Source: https://context7_llms

Provides an example of connecting to a local MCP (Model Context Protocol) server using the stdio transport in Python. This method is suitable for MCP servers running as subprocesses. It demonstrates connecting, listing available tools, and calling a tool.

```python
from blackgeorge.tools import MCPToolProvider

async with MCPToolProvider() as provider:
    await provider.connect_stdio("uv", ["run", "my-mcp-server"])
    tools = provider.list_tools()
    result = await provider.acall_tool("fetch", {"url": "https://example.com"})
```

--------------------------------

### Integration with Splunk

Source: https://jolovicdev.github.io/blackgeorge/logging/index

Example of logging transaction details compatible with Splunk.

```APIDOC
### Splunk

```python
logger.info("Transaction completed", transaction_id="abc123", amount=99.99)
```
```

--------------------------------

### Integration with ELK Stack

Source: https://jolovicdev.github.io/blackgeorge/logging/index

Example of logging API requests compatible with the ELK Stack.

```APIDOC
### ELK Stack (Elasticsearch, Logstash, Kibana)

```python
logger.info("API request", method="GET", path="/api/users", status=200)
```
```

--------------------------------

### Logging within Blackgeorge Custom Stores

Source: https://jolovicdev.github.io/blackgeorge/logging/index

Demonstrates how to incorporate structured logging within a custom store implementation that extends Blackgeorge's `RunStore`. The example logs the creation of a run with debug and info messages.

```python
from blackgeorge.store import RunStore
from blackgeorge.logging import get_logger

logger = get_logger("stores").with_context(component="custom_store")

class CustomRunStore(RunStore):
    def create_run(self, run_id: str, input_payload: dict) -> None:
        logger.debug("Creating run", run_id=run_id)
        # Implementation
        logger.info("Run created", run_id=run_id)
```

--------------------------------

### Python Tool with Combined Pre and Post Hooks

Source: https://context7_llms

Shows a tool decorated with both a pre-hook (`before`) and a post-hook (`after`). The pre-hook logs the start of the tool execution, and the post-hook logs its completion and result. Requires `blackgeorge.tools`.

```python
from blackgeorge.tools import tool

def before(call):
    print(f"Starting {call.name}")

def after(call, result):
    print(f"Finished {call.name}: {result.content}")

@tool(pre=(before,), post=(after,))
def multiply(a: int, b: int) -> int:
    return a * b
```

--------------------------------

### Error Handling in Pre-Hooks (Python)

Source: https://context7_llms

Illustrates how exceptions raised in pre-hooks prevent the tool from executing. This example uses a validation hook to ensure an input value is non-negative.

```python
from blackgeorge.tools import tool

def strict_validation(call):
    if call.arguments.get("value", 0) < 0:
        raise ValueError("Value must be non-negative")

@tool(pre=(strict_validation,))
def process(value: int) -> int:
    return value * 2

# This will fail with ValueError if value < 0
```

--------------------------------

### Get Structured Output with Pydantic Models

Source: https://context7_llms

Demonstrates how to enforce structured output from an agent using Pydantic models. Blackgeorge integrates with Instructor to validate the model's response against the defined schema.

```python
from pydantic import BaseModel

from blackgeorge import Desk, Job, Worker

class Summary(BaseModel):
    title: str
    bullets: list[str]

worker = Worker(name="Writer", model="openai/gpt-5-nano")
desk = Desk(model="openai/gpt-5-nano")
job = Job(input="Summarize the report", response_schema=Summary)

report = desk.run(worker, job)
print(report.data)
```

--------------------------------

### Connect to MCP Server via SSE Transport in Python

Source: https://context7_llms

Illustrates connecting to an MCP server that uses Server-Sent Events (SSE) transport in Python. This method is suitable for MCP servers that push updates using SSE. The example covers establishing the connection and listing available tools.

```python
from blackgeorge.tools import MCPToolProvider

async with MCPToolProvider() as provider:
    await provider.connect_sse("https://api.example.com/mcp/sse")
    tools = provider.list_tools()
```

--------------------------------

### Log Filtering by Level in Blackgeorge

Source: https://jolovicdev.github.io/blackgeorge/logging/index

Shows how to filter logs based on their level using the `level` parameter when getting a logger. This example sets the level to `WARNING`, so only warning, error, and critical messages are outputted.

```python
import logging

from blackgeorge.logging import get_logger

# Only WARNING and above will be logged
logger = get_logger("my_app", level=logging.WARNING)

logger.debug("This won't be logged")
logger.info("Neither will this")
logger.warning("This will be logged")
logger.error("So will this")
```

--------------------------------

### Initialize Instructor Client with LiteLLM Provider

Source: https://context7_llms

Instructor clients can be initialized to work with LiteLLM providers for structured output. This involves using `instructor.from_provider` with the appropriate LiteLLM model string, supporting both synchronous and asynchronous clients.

```python
# Synchronous client
instructor.from_provider("litellm/<model>")

# Asynchronous client
instructor.from_provider("litellm/<model>", async_client=True)
```

--------------------------------

### Create a StructuredLogger Instance

Source: https://jolovicdev.github.io/blackgeorge/logging/index

Demonstrates how to initialize a StructuredLogger with a specific name and logging level. This is the first step to using the structured logging capabilities.

```python
from blackgeorge.logging import StructuredLogger
import logging

logger = StructuredLogger("my_app", level=logging.INFO)
```

--------------------------------

### Create a Desk Instance

Source: https://context7_llms

Initializes a Desk object, which is the main interface for interacting with LLM models. Key parameters control model behavior, token limits, and output structuring.

```python
from blackgeorge import Desk

desk = Desk(
    model="openai/gpt-5-nano",
    temperature=0.2,
    max_tokens=800,
    stream=False,
    structured_output_retries=1,
    max_iterations=10,
    max_tool_calls=20,
    respect_context_window=True,
)
```

--------------------------------

### Caching with Pre and Post-Hooks in Python

Source: https://context7_llms

Illustrates how to implement a caching mechanism using pre-hooks to check the cache and post-hooks to store results. This optimizes performance for expensive computations by avoiding redundant calculations.

```python
from functools import lru_cache
from blackgeorge.tools import tool

cache = {}

def cache_pre(call):
    key = (call.name, frozenset(call.arguments.items()))
    if key in cache:
        # Return cached result
        return cache[key]

def cache_post(call, result):
    key = (call.name, frozenset(call.arguments.items()))
    cache[key] = result

@tool(pre=(cache_pre,), post=(cache_post,))
def expensive_computation(n: int) -> int:
    return sum(range(n))
```

--------------------------------

### Integrating Channel and Blackboard with Workforce in Python

Source: https://context7_llms

Shows how to initialize a Workforce in 'collaborate' mode, passing instances of Channel and Blackboard for inter-worker communication and shared state management.

```python
from blackgeorge import Worker, Workforce
from blackgeorge.collaboration import Channel, Blackboard

channel = Channel()
blackboard = Blackboard()

workforce = Workforce(
    [w1, w2],
    mode="collaborate",
    channel=channel,
    blackboard=blackboard,
)

# Workers can access workforce.channel and workforce.blackboard
```

--------------------------------

### Get WorkerSession History

Source: https://jolovicdev.github.io/blackgeorge/session/index

Retrieves the complete conversation history for a WorkerSession. Each message includes its role (e.g., 'user', 'assistant') and content.

```python
messages = session.history()
for message in messages:
    print(f"{message.role}: {message.content}")
```

--------------------------------

### Subscribe to Desk Events

Source: https://context7_llms

Allows observing events emitted by the Desk, such as run starts and component events. Events can be subscribed to via the `desk.event_bus`.

```python
from blackgeorge import Desk

def handle_event(event) -> None:
    print(event.type, event.source)

desk = Desk(model="openai/gpt-5-nano")
desk.event_bus.subscribe("run.started", handle_event)
```

--------------------------------

### Filter Events by Payload Content

Source: https://context7_llms

Illustrates filtering events based on the content of their payload. This example checks for the presence of 'worker_name' and specifically handles events from the 'analyst' worker.

```python
def handle_worker_events(event):
    if "worker_name" in event.payload:
        worker = event.payload["worker_name"]
        if worker == "analyst":
            print(f"Analyst event: {event.type}")

bus.subscribe("worker.started", handle_worker_events)
```

--------------------------------

### Subscribe to Events using Desk

Source: https://context7_llms

Demonstrates how to subscribe to specific events like 'run.started' and 'tool.completed' using the Desk object's event bus. This is the most common way to integrate event handling into your workflow.

```python
from blackgeorge import Desk

def on_run_started(event):
    print(f"Run {event.run_id} started")

def on_tool_completed(event):
    tool_name = event.payload.get("tool_name")
    print(f"Tool {tool_name} completed")

desk = Desk(model="openai/gpt-5-nano")
desk.event_bus.subscribe("run.started", on_run_started)
desk.event_bus.subscribe("tool.completed", on_tool_completed)
```

--------------------------------

### Shell Command to Set API Key

Source: https://context7_llms

Sets the DEEPSEEK_API_KEY environment variable in the shell. This is a prerequisite for running the coding agent example, allowing it to authenticate with the specified model provider.

```bash
export DEEPSEEK_API_KEY="..."
```

--------------------------------

### Configure Blackgeorge Desk Instance

Source: https://context7_llms

The Desk class in Blackgeorge accepts numerous configuration parameters for model settings, streaming, structured output, execution limits, context window handling, custom components, and storage.

```python
from blackgeorge import Desk

desk = Desk(
    # Model configuration
    model="openai/gpt-5-nano",
    temperature=0.2,
    max_tokens=800,

    # Streaming
    stream=False,

    # Structured output
    structured_output_retries=1,

    # Limits
    max_iterations=10,
    max_tool_calls=20,

    # Context window handling
    respect_context_window=True,

    # Custom components
    event_bus=custom_event_bus,
    run_store=custom_run_store,
    memory_store=custom_memory_store,
    adapter=custom_adapter,

    # Storage
    storage_dir=".blackgeorge",
)
```

--------------------------------

### Filter Events by Type

Source: https://context7_llms

Shows how to subscribe to all events using a wildcard '*' and then filter them within the handler based on their type, specifically targeting 'tool.' events in this example. This allows for centralized event processing.

```python
def handle_all_events(event):
    if event.type.startswith("tool."):
        print(f"Tool event: {event.type}")
        # Process tool events

bus.subscribe("*", handle_all_events)
```

--------------------------------

### Initialize Toolbelt for Tool Management in Python

Source: https://context7_llms

Demonstrates the basic initialization of a `Toolbelt` object in Python. `Toolbelt` (also aliased as `Toolkit`) is responsible for managing the registration and lookup of available tools within the application.

```python
from blackgeorge.tools import Toolbelt

belt = Toolbelt()
```

--------------------------------

### Thinking Models with Reasoning Content (DeepSeek)

Source: https://context7_llms

Configures a session to use thinking models, specifically demonstrating the DeepSeek Reasoner. It shows how to enable reasoning and highlights that this model does not support `budget_tokens`.

```python
from blackgeorge import Job

# DeepSeek Reasoner (no budget_tokens support)
report = session.run(
    "Which is larger: 9.11 or 9.8?",
    thinking={"type": "enabled"},
)
```

--------------------------------

### Python Tool Post-hook for Logging Completion Status

Source: https://context7_llms

A post-hook example `log_completion` that logs detailed status messages for tool execution, including timeouts, cancellations, errors, and success. Requires `blackgeorge.tools`.

```python
from blackgeorge.tools import tool

def log_completion(call, result):
    if result.timed_out:
        print(f"WARNING: {call.name} timed out")
    elif result.cancelled:
        print(f"WARNING: {call.name} was cancelled")
    elif result.error:
        print(f"ERROR in {call.name}: {result.error}")
    else:
        print(f"SUCCESS: {call.name} returned {result.content}")

@tool(
    timeout=5.0,
    post=(log_completion,),
)
def fetch_data(url: str) -> str:
    import requests
    return requests.get(url).text
```

--------------------------------

### Channel Creation and Initialization

Source: https://context7_llms

Shows how to instantiate a Channel object, which is essential for enabling direct messaging and broadcasting between workers in a Blackgeorge environment. No external dependencies are required beyond the blackgeorge library.

```python
from blackgeorge.collaboration import Channel

channel = Channel()
```

--------------------------------

### Add a Tool to a Worker

Source: https://context7_llms

Shows how to define and add a custom tool to a Blackgeorge Worker. The '@tool' decorator registers the function, making it available for the agent to call. The example uses an 'echo' tool.

```python
from blackgeorge import Desk, Job, Worker
from blackgeorge.tools import tool

@tool(description="Echo text back")
def echo(text: str) -> str:
    return text

desk = Desk(model="openai/gpt-5-nano")
worker = Worker(name="Agent", tools=[echo])
job = Job(input="Call echo with text=hello")

report = desk.run(worker, job)
print(report.content)
```

--------------------------------

### Connect to Remote MCP Server with Authentication in Python

Source: https://context7_llms

Demonstrates connecting to a remote MCP server via streamable HTTP in Python when authentication is required. It shows how to create a custom `httpx.AsyncClient` with necessary headers, such as authorization tokens, and pass it to the `MCPToolProvider`.

```python
import httpx
from blackgeorge.tools import MCPToolProvider

async with MCPToolProvider() as provider:
    client = httpx.AsyncClient(headers={"Authorization": "Bearer token"})
    await provider.connect_streamable_http("https://api.example.com/mcp", http_client=client)
    tools = provider.list_tools()
```

--------------------------------

### Build Documentation Site

Source: https://context7_llms

Builds the static HTML files for the MkDocs documentation site. This command generates the final documentation that can be deployed.

```text
uv run mkdocs build
```

--------------------------------

### Get a Blackgeorge Logger

Source: https://jolovicdev.github.io/blackgeorge/logging/index

Demonstrates how to obtain a logger instance from the Blackgeorge library. This logger is used for all subsequent logging operations. It requires a name for the logger, typically representing the application or module.

```python
from blackgeorge.logging import get_logger

logger = get_logger("my_app")
logger.info("Application started")
```

--------------------------------

### Async Python Tool with Async Pre and Post Hooks

Source: https://context7_llms

Demonstrates using asynchronous pre and post hooks with an asynchronous tool. Both hooks simulate async work using `asyncio.sleep` before proceeding. Requires `asyncio` and `blackgeorge.tools`.

```python
import asyncio
from blackgeorge.tools import tool

async def async_pre_hook(call):
    await asyncio.sleep(0.1)  # Simulate async work
    print(f"Async pre-hook for {call.name}")

async def async_post_hook(call, result):
    await asyncio.sleep(0.1)
    print(f"Async post-hook for {call.name}")

@tool(pre=(async_pre_hook,), post=(async_post_hook,))
async def async_tool(value: str) -> str:
    await asyncio.sleep(0.1)
    return value.upper()
```

--------------------------------

### Channel: Receiving Messages (Default Behavior)

Source: https://context7_llms

Explains how a worker can receive messages intended for it from the Channel. By default, messages are cleared from the channel after being read. The example shows iterating through received messages and printing their content.

```python
# Receive all messages for this worker (clears after reading)
messages = channel.receive("worker_b")
for msg in messages:
    print(f"From {msg.sender}: {msg.content}")
```

--------------------------------

### Using Thinking Models for Reasoning

Source: https://jolovicdev.github.io/blackgeorge/session/index

Illustrates how to enable 'thinking' models for LLM reasoning. This allows the model to output its thought process before providing the final answer. Supports `budget_tokens` for specific models like Claude 3.7+.

```python
from blackgeorge import Job

# DeepSeek Reasoner (no budget_tokens support)
report = session.run(
    "Which is larger: 9.11 or 9.8?",
    thinking={"type": "enabled"},
)

# Anthropic Claude 3.7+ (supports budget_tokens)
report = session.run(
    "Which is larger: 9.11 or 9.8?",
    thinking={"type": "enabled", "budget_tokens": 1024},
)

print(f"Reasoning: {report.reasoning_content}")
print(f"Answer: {report.content}")
```

--------------------------------

### Create and Use EventBus for Event Subscription and Emission

Source: https://context7_llms

This Python code illustrates how to initialize an EventBus, subscribe handler functions to specific event types (synchronously and asynchronously), and emit events. It covers both synchronous and asynchronous emission, detailing the structure of the 'Event' object and providing a strategy for unsubscribing by managing handler state.

```python
from blackgeorge.event_bus import EventBus
from blackgeorge.core.event import Event
from datetime import datetime

bus = EventBus()

def handle_event(event) -> None:
    print(f"{event.type}: {event.payload}")

# Subscribe to a specific event type
bus.subscribe("run.started", handle_event)

# Subscribe to multiple event types
bus.subscribe("run.completed", handle_event)
bus.subscribe("run.failed", handle_event)

# Synchronous event emission
event_sync = Event(
    type="custom.event",
    source="my_component",
    run_id="run-123",
    payload={"data": "value"}
)
bus.emit(event_sync)

# Asynchronous event emission (assuming an async context or event loop)
async def async_emit_example():
    event_async = Event(
        type="async.event",
        source="async_component",
        run_id="run-456",
        payload={"async_data": "async_value"}
    )
    await bus.aemit(event_async)

# To run the async part, you would typically use:
# import asyncio
# asyncio.run(async_emit_example())

# Example of unsubscribing by managing handler state
class EventHandler:
    def __init__(self):
        self.enabled = True

    def handle(self, event):
        if not self.enabled:
            return
        print(f"Conditional handler: {event.type}")

handler = EventHandler()
bus.subscribe("run.started", handler.handle)

# To stop handling events later:
# handler.enabled = False

```

--------------------------------

### Running Synchronous and Asynchronous Conversations with WorkerSession in Python

Source: https://context7_llms

Provides examples of sending messages to a WorkerSession using both synchronous (`run()`) and asynchronous (`arun()`) methods. The session automatically handles message persistence and context.

```python
report1 = session.run("Hello, I'm Alice")
print(report1.content)

report2 = session.run("What's my name?")
print(report2.content)

# For async:
# report1_async = await session.arun("Hello, I'm Alice")
```

--------------------------------

### Create Workers and Workforce with Collaboration Tools

Source: https://context7_llms

This Python snippet demonstrates how to create 'analyst' and 'reviewer' workers with specific collaboration tools (blackboard and channel interactions) and then assemble them into a 'collaborate' mode workforce. It concludes by initializing a 'Desk' and running a 'Job' through the workforce.

```python
from blackgeorge.core.worker import Worker
from blackgeorge.core.workforce import Workforce
from blackgeorge.core.desk import Desk
from blackgeorge.core.job import Job

# Assuming blackboard, channel, and tools are defined elsewhere
# Example placeholder definitions:
class MockBlackboard:
    pass

class MockChannel:
    pass

def blackboard_write_tool(bb, name):
    return f"write_tool_for_{name}"

def channel_send_tool(ch, name):
    return f"send_tool_for_{name}"

def blackboard_read_tool(bb):
    return "read_tool"

def channel_receive_tool(ch, name):
    return f"receive_tool_for_{name}"

blackboard = MockBlackboard()
channel = MockChannel()

# Create workers with collaboration tools
analyst = Worker(
    name="analyst",
    tools=[
        blackboard_write_tool(blackboard, "analyst"),
        channel_send_tool(channel, "analyst"),
    ],
)

reviewer = Worker(
    name="reviewer",
    tools=[
        blackboard_read_tool(blackboard),
        blackboard_write_tool(blackboard, "reviewer"),
        channel_send_tool(channel, "reviewer"),
        channel_receive_tool(channel, "reviewer"),
    ],
)

# Create workforce
workforce = Workforce(
    [analyst, reviewer],
    mode="collaborate",
    channel=channel,
    blackboard=blackboard,
)

# Run
desk = Desk(model="openai/gpt-5-nano")
job = Job(input="Analyze the data and have it reviewed")
report = desk.run(workforce, job)

```

--------------------------------

### Inspecting Tool Results with Post-Hooks in Python

Source: https://context7_llms

Demonstrates how to use a post-hook to inspect detailed information about a tool's execution result, including content, data, errors, and status flags like timeout or cancellation. This is useful for debugging and monitoring.

```python
from blackgeorge.tools import tool, ToolResult

def inspect_result(call, result):
    print(f"Tool: {call.name}")
    print(f"Content: {result.content}")
    print(f"Data: {result.data}")
    print(f"Error: {result.error}")
    print(f"Timed out: {result.timed_out}")
    print(f"Cancelled: {result.cancelled}")

@tool(timeout=10.0, post=(inspect_result,))
async def long_running_task(seconds: int) -> str:
    import asyncio
    await asyncio.sleep(seconds)
    return f"Slept for {seconds} seconds"
```

--------------------------------

### Log API Requests for ELK Stack

Source: https://jolovicdev.github.io/blackgeorge/logging/index

An example of logging an API request with relevant details such as HTTP method, path, and status code. This structured log entry is compatible with the ELK Stack for analysis.

```python
logger.info("API request", method="GET", path="/api/users", status=200)
```

--------------------------------

### Implement Multi-Component Logging with Shared Context

Source: https://jolovicdev.github.io/blackgeorge/logging/index

Demonstrates how to create a base logger with application-wide context (like app name and version) and then derive component-specific loggers. All logs from these derived loggers will automatically include the shared context.

```python
from blackgeorge.logging import get_logger

# Create base logger for the application
base_logger = get_logger("myapp").with_context(
    app="myapp",
    version="1.0.0"
)

# Create component-specific loggers
db_logger = base_logger.with_context(component="database")
api_logger = base_logger.with_context(component="api")
worker_logger = base_logger.with_context(component="worker")

# All logs include app and version context
db_logger.info("Connected to database")
api_logger.info("Received request")
worker_logger.info("Processing job")
```

--------------------------------

### Implement Argument Validation Pre-hook in Python

Source: https://context7_llms

Shows a practical example of a pre-hook in Python used for argument validation. The `validate_positive` function checks if arguments are positive numbers before the decorated tool function is executed, raising a `ValueError` if the condition is not met.

```python
from blackgeorge.tools import tool

def validate_positive(call):
    for key, value in call.arguments.items():
        if isinstance(value, (int, float)) and value < 0:
            raise ValueError(f"{key} must be positive, got {value}")

@tool(pre=(validate_positive,))
def square_root(x: float) -> float:
    return x ** 0.5
```

--------------------------------

### Blackboard Write Tool Wrapper

Source: https://context7_llms

Offers a tool wrapper for `blackboard.write`, allowing workers to store key-value pairs in the Blackboard using tool calls. Initialization requires the Blackboard instance and the author's name.

```python
from blackgeorge.collaboration import blackboard_write_tool

write_tool = blackboard_write_tool(blackboard, author="worker_a")

worker = Worker(name="worker_a", tools=[write_tool])
```

--------------------------------

### Logging within Blackgeorge Event Handlers

Source: https://jolovicdev.github.io/blackgeorge/logging/index

Illustrates logging within an event handler function subscribed to a Blackgeorge event bus. The example logs when a tool completes, capturing the tool name and run ID from the event payload.

```python
from blackgeorge import Desk
from blackgeorge.logging import get_logger

logger = get_logger("events").with_context(component="event_monitor")

def on_tool_completed(event):
    tool_name = event.payload.get("tool_name")
    logger.info("Tool completed", tool=tool_name, run_id=event.run_id)

desk = Desk(model="openai/gpt-5-nano")
desk.event_bus.subscribe("tool.completed", on_tool_completed)
```

--------------------------------

### Subscribing to Blackboard Changes in Python

Source: https://context7_llms

Explains how to set up event listeners for Blackboard updates. You can subscribe to changes for a specific key or to all changes across the blackboard, enabling real-time reaction to shared state modifications.

```python
def on_write(key: str, value: Any, author: str) -> None:
    print(f"{author} wrote {key}: {value}")

# Subscribe to a specific key
blackboard.subscribe("analysis_result", on_write)

# Subscribe to all changes
blackboard.subscribe_all(on_write)
```

--------------------------------

### Connect to Remote MCP Server via Streamable HTTP in Python

Source: https://context7_llms

Shows how to connect to a remote MCP server using the streamable HTTP transport in Python. This is used for MCP servers accessible over HTTP. The example includes connecting, listing tools, and calling a tool.

```python
from blackgeorge.tools import MCPToolProvider

async with MCPToolProvider() as provider:
    await provider.connect_streamable_http("https://api.example.com/mcp")
    tools = provider.list_tools()
    result = await provider.acall_tool("search", {"query": "python"})
```

--------------------------------

### Reading from a Blackboard in Python

Source: https://context7_llms

Shows how to retrieve data from a Blackboard using a key. It also demonstrates how to check for the existence of a key before attempting to read it, ensuring safe data access.

```python
result = blackboard.read("analysis_result")
print(result)  # {"score": 95}

# Check if a key exists
if blackboard.exists("analysis_result"):
    value = blackboard.read("analysis_result")
```

--------------------------------

### Multi-step Task Use Case: Tracking Progress

Source: https://context7_llms

Shows how to manage multi-step tasks by tracking progress within a session. Metadata, including the current step and total steps, is stored with the session, allowing for progress monitoring across interactions.

```python
session = desk.session(
    worker=planner_worker,
    session_id=f"plan:{plan_id}",
    metadata={"step": 1, "total_steps": 5},
)
```

--------------------------------

### Configure Blackgeorge Workforce

Source: https://context7_llms

This Python code shows how to set up a Blackgeorge Workforce, specifying the workers, coordination mode, and optional manager or reducer functions.

```python
from blackgeorge import Worker, Workforce

workers = [Worker(name="a"), Worker(name="b")]

workforce = Workforce(
    workers=workers,
    mode="managed",           # "managed" or "collaborate"
    name="team",
    manager=manager_worker,   # Only for managed mode
    reducer=custom_reducer,   # Only for collaborate mode
    channel=channel,          # Optional
    blackboard=blackboard,    # Optional
)
```

--------------------------------

### Channel Broadcast Tool Wrapper

Source: https://context7_llms

Presents a tool wrapper for `channel.broadcast`, allowing workers to broadcast messages using tool calls. Initialization requires the Channel instance and the sender's name.

```python
from blackgeorge.collaboration import channel_broadcast_tool

broadcast_tool = channel_broadcast_tool(channel, sender="manager")

worker = Worker(name="worker_a", tools=[broadcast_tool])
```

--------------------------------

### Retry-Aware Pre-Hooks in Python

Source: https://context7_llms

Shows how to implement a pre-hook that tracks the number of attempts for a tool, useful for operations that might fail intermittently. The tool is configured with a retry count.

```python
from blackgeorge.tools import tool

attempt_count = {}

def count_attempts(call):
    attempt_count[call.id] = attempt_count.get(call.id, 0) + 1
    print(f"Attempt {attempt_count[call.id]} for {call.name}")

@tool(retries=3, pre=(count_attempts,))
def flaky_operation(value: str) -> str:
    import random
    if random.random() < 0.5:
        raise Exception("Random failure")
    return value
```

--------------------------------

### Thinking Models with Reasoning Content (Claude)

Source: https://context7_llms

Demonstrates using thinking models with Claude 3.7+ (or similar) which supports `budget_tokens`. This allows for controlling the token budget allocated for reasoning, alongside enabling the thinking process.

```python
# Anthropic Claude 3.7+ (supports budget_tokens)
report = session.run(
    "Which is larger: 9.11 or 9.8?",
    thinking={"type": "enabled", "budget_tokens": 1024},
)
```

--------------------------------

### Track Tool Execution with Structured Logging

Source: https://jolovicdev.github.io/blackgeorge/logging/index

Shows how to use structured logging within a tool function to track its execution, including the SQL query being run, the number of rows affected, and any errors encountered. This is useful for debugging complex operations.

```python
from blackgeorge.tools import tool
from blackgeorge.logging import get_logger

logger = get_logger("tools").with_context(component="database")

@tool()
def query_database(sql: str) -> list[dict]:
    logger.info("Executing query", sql=sql)
    try:
        results = execute_sql(sql)
        logger.info("Query completed", row_count=len(results))
        return results
    except Exception as e:
        logger.error("Query failed", sql=sql, error=str(e))
        raise
```

--------------------------------

### Blackboard for Shared State in Python

Source: https://context7_llms

Illustrates the Blackboard class for sharing state across workers. Shows how to write data, read data, and subscribe to changes in shared keys.

```python
from blackgeorge.collaboration import Blackboard

bb = Blackboard()

bb.write("analysis_result", {"score": 95}, author="analyst")

result = bb.read("analysis_result")

bb.subscribe("analysis_result", lambda k, v, a: print(f"Updated by {a}"))
```

--------------------------------

### Logging Various Data Types

Source: https://jolovicdev.github.io/blackgeorge/logging/index

Demonstrates how to log different data types using the logger.

```APIDOC
## Data types

The logger handles various data types in context:

```python
logger.info(
    "Complex data",
    integer=42,
    floating=3.14,
    boolean=True,
    none_value=None,
    list=[1, 2, 3],
    dict={"key": "value"},
    custom_object=str(CustomClass())
)
```
```

--------------------------------

### Configure Blackgeorge Worker

Source: https://context7_llms

This Python code demonstrates how to initialize a Blackgeorge Worker with various configuration options including identity, model, tools, and memory scope.

```python
from blackgeorge import Worker

worker = Worker(
    # Identity
    name="Analyst",

    # Model
    model="openai/gpt-4-turbo",
    instructions="You are a data analyst.",

    # Tools
    tools=[tool1, tool2],

    # Memory
    memory_scope="analyst:session-1",
)
```