diff --git a/hooks.md b/hooks.md
new file mode 100644
index 0000000..b902232
--- /dev/null
+++ b/hooks.md
@@ -0,0 +1,235 @@
+# Hooks System Documentation
+
+This hooks system allows you to define hooks at both the client level and per individual `create` call, providing maximum flexibility for logging, metrics, debugging, and other cross-cutting concerns.
+
+## Overview
+
+The hooks system consists of two main classes:
+
+- **`Hooks`**: A container for hook functions that can be executed with metadata
+- **`Client`**: A base client class that supports both global and call-specific hooks
+
+## Basic Usage
+
+### 1. Creating a Hooks Object
+
+```python
+from hooks import Hooks
+
+# Create a hooks container
+call_hooks = Hooks()
+
+# Add hook functions
+def my_hook(**kwargs):
+    print(f"Hook called with: {kwargs}")
+
+call_hooks.add_hook(my_hook)
+```
+
+### 2. Using Hooks with a Client
+
+```python
+from hooks import Client, Hooks
+
+class MyClient(Client):
+    def _create_impl(self, data, hooks=None, **kwargs):
+        # Your implementation here
+        result = f"processed: {data}"
+        
+        # Call hooks with metadata
+        if hooks:
+            hooks.meta(input=data, output=result)
+        
+        return result
+
+# Create client and add global hooks
+client = MyClient()
+client.hooks.add_hook(lambda **kwargs: print(f"Global: {kwargs}"))
+
+# Create call-specific hooks
+call_hooks = Hooks()
+call_hooks.add_hook(lambda **kwargs: print(f"Call: {kwargs}"))
+
+# Use both types of hooks
+result = client.create("some data", hooks=call_hooks)
+```
+
+## Key Features
+
+### 1. Client-Level Hooks
+
+Hooks defined at the client level are executed for every `create` call:
+
+```python
+client = MyClient()
+client.hooks.add_hook(logging_hook)
+client.hooks.add_hook(metrics_hook)
+
+# These hooks will run for all create calls
+result1 = client.create("query 1")
+result2 = client.create("query 2")
+```
+
+### 2. Call-Specific Hooks
+
+You can add hooks for individual `create` calls:
+
+```python
+call_hooks = Hooks()
+call_hooks.add_hook(debug_hook)
+
+# This debug hook only runs for this specific call
+result = client.create("special query", hooks=call_hooks)
+```
+
+### 3. Combined Hooks
+
+When both client-level and call-specific hooks are present, all hooks are executed:
+
+```python
+# Client has global logging hook
+client.hooks.add_hook(logging_hook)
+
+# Add call-specific debug hook
+call_hooks = Hooks()
+call_hooks.add_hook(debug_hook)
+
+# Both logging_hook and debug_hook will be executed
+result = client.create("query", hooks=call_hooks)
+```
+
+## Hook Function Signature
+
+Hook functions should accept keyword arguments:
+
+```python
+def my_hook(**kwargs):
+    # Access metadata
+    input_data = kwargs.get('input')
+    output_data = kwargs.get('output')
+    client_name = kwargs.get('client')
+    
+    # Your hook logic here
+    print(f"Processing {input_data} -> {output_data}")
+```
+
+## Common Hook Patterns
+
+### Logging Hook
+
+```python
+def logging_hook(**kwargs):
+    print(f"[LOG] {kwargs}")
+```
+
+### Metrics Hook
+
+```python
+def metrics_hook(**kwargs):
+    input_len = len(str(kwargs.get('input', '')))
+    output = kwargs.get('output', 'N/A')
+    print(f"[METRICS] Input length: {input_len}, Output: {output}")
+```
+
+### Debug Hook
+
+```python
+def debug_hook(**kwargs):
+    client = kwargs.get('client', 'unknown')
+    input_data = kwargs.get('input')
+    print(f"[DEBUG] Client '{client}' processed: {input_data}")
+```
+
+## Integration with Braintrust-Style Evaluations
+
+This hooks system is designed to work seamlessly with Braintrust-style evaluation patterns:
+
+```python
+# Your existing task function pattern
+async def task(query, hooks):
+    # Process the query
+    result = await process_query(query)
+    
+    # Call hooks.meta() as before
+    hooks.meta(input=query, output=result)
+    
+    return result
+
+# Now you can also use the client pattern
+client = EvaluationClient()
+client.hooks.add_hook(global_logging_hook)
+
+call_hooks = Hooks()
+call_hooks.add_hook(specific_debug_hook)
+
+result = await client.create(query, hooks=call_hooks)
+```
+
+## Advanced Usage
+
+### Custom Client Implementation
+
+```python
+class MyEvaluationClient(Client):
+    def __init__(self, name):
+        super().__init__()
+        self.name = name
+    
+    async def _create_impl(self, query, hooks=None, **kwargs):
+        # Your custom logic here
+        result = await self.process_query(query)
+        
+        # Call hooks with custom metadata
+        if hooks:
+            hooks.meta(
+                client=self.name,
+                input=query,
+                output=result,
+                timestamp=time.time(),
+                **kwargs
+            )
+        
+        return result
+    
+    async def create(self, query, hooks=None, **kwargs):
+        combined_hooks = self.hooks.combine_with(hooks)
+        return await self._create_impl(query, hooks=combined_hooks, **kwargs)
+```
+
+### Hook Composition
+
+```python
+# Create reusable hook combinations
+def create_standard_hooks():
+    hooks = Hooks()
+    hooks.add_hook(logging_hook)
+    hooks.add_hook(metrics_hook)
+    return hooks
+
+# Use in different contexts
+call_hooks = create_standard_hooks()
+call_hooks.add_hook(custom_debug_hook)
+
+result = client.create("query", hooks=call_hooks)
+```
+
+## Benefits
+
+1. **Simple**: Easy to understand and implement
+2. **Flexible**: Support both global and per-call hooks
+3. **Composable**: Hooks can be combined and reused
+4. **Compatible**: Works with existing Braintrust patterns
+5. **Extensible**: Easy to add new hook types and metadata
+
+## Example Output
+
+When running with multiple hooks:
+
+```
+[LOG] {'client': 'my-client', 'input': 'test query', 'output': 'processed: test query'}
+[METRICS] Input length: 10, Output: processed: test query
+[DEBUG] Client 'my-client' processed input: test query
+Result: processed: test query
+```
+
+This shows all three hooks (logging, metrics, debug) being executed in order with the same metadata.
\ No newline at end of file
diff --git a/hooks.py b/hooks.py
new file mode 100644
index 0000000..958aec3
--- /dev/null
+++ b/hooks.py
@@ -0,0 +1,116 @@
+"""
+Simple hooks implementation that allows defining hooks at the client level
+and passing additional hooks on individual create calls.
+"""
+
+class Hooks:
+    """A simple hooks container that can store and execute hook functions."""
+    
+    def __init__(self):
+        self._hooks = []
+    
+    def add_hook(self, hook_func):
+        """Add a hook function to be executed."""
+        self._hooks.append(hook_func)
+    
+    def meta(self, **kwargs):
+        """Execute all hooks with the provided metadata."""
+        for hook in self._hooks:
+            # Call each hook function with the metadata
+            hook(**kwargs)
+    
+    def combine_with(self, other_hooks):
+        """Combine this hooks object with another hooks object."""
+        combined = Hooks()
+        combined._hooks = self._hooks.copy()
+        if other_hooks:
+            combined._hooks.extend(other_hooks._hooks)
+        return combined
+
+
+class Client:
+    """A client that supports hooks at the client level and per-create call."""
+    
+    def __init__(self):
+        self.hooks = Hooks()
+    
+    def create(self, *args, hooks=None, **kwargs):
+        """
+        Create method that combines client-level hooks with call-level hooks.
+        
+        Args:
+            *args: Arguments to pass to the actual create implementation
+            hooks: Optional Hooks object for this specific create call
+            **kwargs: Keyword arguments to pass to the actual create implementation
+            
+        Returns:
+            The result of the create operation
+        """
+        # Combine client-level hooks with call-level hooks
+        combined_hooks = self.hooks.combine_with(hooks)
+        
+        # Replace the hooks parameter with the combined hooks
+        kwargs['hooks'] = combined_hooks
+        
+        # Call the actual create implementation
+        return self._create_impl(*args, **kwargs)
+    
+    def _create_impl(self, *args, **kwargs):
+        """
+        Override this method in subclasses to implement the actual create logic.
+        The hooks will be available in kwargs['hooks'].
+        """
+        raise NotImplementedError("Subclasses must implement _create_impl")
+
+
+# Example usage and testing
+if __name__ == "__main__":
+    # Example implementation
+    class ExampleClient(Client):
+        def _create_impl(self, data, hooks=None, **kwargs):
+            print(f"Creating with data: {data}")
+            
+            # Execute hooks with some metadata
+            if hooks:
+                hooks.meta(
+                    input=data,
+                    output="created_successfully",
+                    operation="create"
+                )
+            
+            return "created_successfully"
+    
+    # Define some hook functions
+    def client_hook(**kwargs):
+        print(f"Client hook called with: {kwargs}")
+    
+    def call_hook(**kwargs):
+        print(f"Call hook called with: {kwargs}")
+    
+    # Test the implementation
+    print("Testing hooks implementation:")
+    print("=" * 50)
+    
+    # Create client and add a client-level hook
+    client = ExampleClient()
+    client.hooks.add_hook(client_hook)
+    
+    # Create a hooks object for a single call
+    call_hooks = Hooks()
+    call_hooks.add_hook(call_hook)
+    
+    # Test 1: Create with both client and call hooks
+    print("\nTest 1: Both client and call hooks")
+    result = client.create({"key": "value"}, hooks=call_hooks)
+    print(f"Result: {result}")
+    
+    # Test 2: Create with only client hooks
+    print("\nTest 2: Only client hooks")
+    result = client.create({"key": "value2"})
+    print(f"Result: {result}")
+    
+    # Test 3: Create with only call hooks (no client hooks)
+    print("\nTest 3: Only call hooks (clean client)")
+    client2 = ExampleClient()
+    result = client2.create({"key": "value3"}, hooks=call_hooks)
+    print(f"Result: {result}")
\ No newline at end of file
diff --git a/hooks_example.py b/hooks_example.py
new file mode 100644
index 0000000..c5fee14
--- /dev/null
+++ b/hooks_example.py
@@ -0,0 +1,148 @@
+"""
+Example showing how to use the hooks system with a Braintrust-style evaluation pattern.
+This demonstrates the requested functionality where hooks can be defined at the client level
+and also passed to individual create calls.
+"""
+
+from hooks import Hooks, Client
+import asyncio
+
+
+class EvaluationClient(Client):
+    """
+    Example client that demonstrates hooks integration with evaluation tasks.
+    This simulates how you might use hooks with Braintrust-style evaluations.
+    """
+    
+    def __init__(self, name="evaluation-client"):
+        super().__init__()
+        self.name = name
+    
+    async def _create_impl(self, query, hooks=None, **kwargs):
+        """
+        Simulate an evaluation task that processes a query and calls hooks.
+        This is similar to the task functions used with Braintrust EvalAsync.
+        """
+        # Simulate some processing
+        result = f"processed: {query}"
+        
+        # Call hooks with metadata (similar to Braintrust hooks.meta())
+        if hooks:
+            hooks.meta(
+                client=self.name,
+                input=query,
+                output=result,
+                **kwargs
+            )
+        
+        return result
+    
+    async def create(self, query, hooks=None, **kwargs):
+        """Async version of create method."""
+        # Combine client-level hooks with call-level hooks
+        combined_hooks = self.hooks.combine_with(hooks)
+        
+        # Call the implementation with combined hooks
+        return await self._create_impl(query, hooks=combined_hooks, **kwargs)
+
+
+# Example hook functions
+def logging_hook(**kwargs):
+    """Hook that logs all metadata."""
+    print(f"[LOG] {kwargs}")
+
+def metrics_hook(**kwargs):
+    """Hook that tracks metrics."""
+    print(f"[METRICS] Input length: {len(str(kwargs.get('input', '')))}, "
+          f"Output: {kwargs.get('output', 'N/A')}")
+
+def debug_hook(**kwargs):
+    """Hook that prints debug information."""
+    client = kwargs.get('client', 'unknown')
+    print(f"[DEBUG] Client '{client}' processed input: {kwargs.get('input')}")
+
+
+async def main():
+    """Demonstrate the hooks system with various scenarios."""
+    
+    print("Hooks System Example")
+    print("=" * 60)
+    
+    # Scenario 1: Client with global hooks
+    print("\n1. Client with global hooks only:")
+    print("-" * 40)
+    
+    client = EvaluationClient("global-hooks-client")
+    client.hooks.add_hook(logging_hook)
+    client.hooks.add_hook(metrics_hook)
+    
+    result = await client.create("What is the weather today?")
+    print(f"Result: {result}")
+    
+    # Scenario 2: Client with both global and call-specific hooks
+    print("\n2. Client with both global and call-specific hooks:")
+    print("-" * 40)
+    
+    call_hooks = Hooks()
+    call_hooks.add_hook(debug_hook)
+    
+    result = await client.create(
+        "How do I reset my password?", 
+        hooks=call_hooks,
+        metadata="password-reset-query"
+    )
+    print(f"Result: {result}")
+    
+    # Scenario 3: Multiple calls with different call-specific hooks
+    print("\n3. Multiple calls with different call-specific hooks:")
+    print("-" * 40)
+    
+    # First call with debug hook
+    call_hooks1 = Hooks()
+    call_hooks1.add_hook(debug_hook)
+    
+    result1 = await client.create("Book a flight to Paris", hooks=call_hooks1)
+    print(f"Result 1: {result1}")
+    
+    # Second call with no additional hooks (only global ones)
+    result2 = await client.create("Cancel my subscription")
+    print(f"Result 2: {result2}")
+    
+    # Scenario 4: New client with only call-specific hooks
+    print("\n4. New client with only call-specific hooks:")
+    print("-" * 40)
+    
+    client2 = EvaluationClient("call-hooks-only-client")
+    
+    call_hooks2 = Hooks()
+    call_hooks2.add_hook(logging_hook)
+    call_hooks2.add_hook(debug_hook)
+    
+    result = await client2.create("What are my account settings?", hooks=call_hooks2)
+    print(f"Result: {result}")
+    
+    # Scenario 5: Simulating the pattern from the user's request
+    print("\n5. User's requested pattern:")
+    print("-" * 40)
+    
+    # This demonstrates the exact pattern the user wanted:
+    # call_hooks = Hooks()
+    # call_hooks...
+    # client.create(..., hooks=call_hooks)
+    
+    client3 = EvaluationClient("user-pattern-client")
+    client3.hooks.add_hook(logging_hook)  # Global hook
+    
+    call_hooks = Hooks()
+    call_hooks.add_hook(metrics_hook)
+    call_hooks.add_hook(debug_hook)
+    
+    result = await client3.create(
+        "Process this important query",
+        hooks=call_hooks
+    )
+    print(f"Result: {result}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
\ No newline at end of file
diff --git a/integration_example.py b/integration_example.py
new file mode 100644
index 0000000..299d7f7
--- /dev/null
+++ b/integration_example.py
@@ -0,0 +1,188 @@
+"""
+Integration example showing how the hooks system works alongside existing Braintrust patterns.
+This demonstrates both the traditional task function approach and the new client approach.
+"""
+
+from hooks import Hooks, Client
+import asyncio
+
+
+# Traditional Braintrust-style task function
+async def traditional_task(query, hooks):
+    """
+    Traditional task function that receives hooks from Braintrust EvalAsync.
+    This is how hooks currently work in the existing codebase.
+    """
+    # Simulate processing
+    result = f"traditional_result: {query}"
+    
+    # Call hooks.meta() as done in existing code
+    hooks.meta(
+        input=query,
+        output=result,
+        approach="traditional"
+    )
+    
+    return result
+
+
+# New client-based approach
+class EvalClient(Client):
+    """Client that can be used alongside traditional Braintrust evaluation."""
+    
+    def __init__(self, name="eval-client"):
+        super().__init__()
+        self.name = name
+    
+    async def _create_impl(self, query, hooks=None, **kwargs):
+        """Implementation that works with the hooks system."""
+        # Simulate processing (same as traditional task)
+        result = f"client_result: {query}"
+        
+        # Call hooks with metadata
+        if hooks:
+            hooks.meta(
+                input=query,
+                output=result,
+                client=self.name,
+                approach="client-based",
+                **kwargs
+            )
+        
+        return result
+    
+    async def create(self, query, hooks=None, **kwargs):
+        """Async create method that combines hooks."""
+        combined_hooks = self.hooks.combine_with(hooks)
+        return await self._create_impl(query, hooks=combined_hooks, **kwargs)
+
+
+# Hook functions for demonstration
+def braintrust_style_hook(**kwargs):
+    """Hook that mimics Braintrust-style logging."""
+    approach = kwargs.get('approach', 'unknown')
+    input_data = kwargs.get('input')
+    output_data = kwargs.get('output')
+    print(f"[BRAINTRUST-STYLE] {approach}: {input_data} -> {output_data}")
+
+
+def custom_metrics_hook(**kwargs):
+    """Custom metrics hook."""
+    client = kwargs.get('client', 'N/A')
+    input_len = len(str(kwargs.get('input', '')))
+    print(f"[METRICS] Client: {client}, Input length: {input_len}")
+
+
+async def demonstrate_integration():
+    """
+    Demonstrate how the new hooks system integrates with existing patterns.
+    """
+    
+    print("Integration Example: Traditional vs Client-based Hooks")
+    print("=" * 60)
+    
+    # Sample query
+    query = "What is the weather in New York?"
+    
+    # 1. Traditional Braintrust-style approach
+    print("\n1. Traditional Braintrust-style task function:")
+    print("-" * 50)
+    
+    # Create hooks object (simulating what Braintrust would pass)
+    traditional_hooks = Hooks()
+    traditional_hooks.add_hook(braintrust_style_hook)
+    
+    result1 = await traditional_task(query, traditional_hooks)
+    print(f"Result: {result1}")
+    
+    # 2. New client-based approach with global hooks
+    print("\n2. New client-based approach with global hooks:")
+    print("-" * 50)
+    
+    client = EvalClient("weather-client")
+    client.hooks.add_hook(braintrust_style_hook)
+    client.hooks.add_hook(custom_metrics_hook)
+    
+    result2 = await client.create(query)
+    print(f"Result: {result2}")
+    
+    # 3. Client-based with both global and call-specific hooks
+    print("\n3. Client with both global and call-specific hooks:")
+    print("-" * 50)
+    
+    # Create call-specific hooks
+    call_hooks = Hooks()
+    call_hooks.add_hook(lambda **kwargs: print(f"[CALL-SPECIFIC] Processing: {kwargs.get('input')}"))
+    
+    result3 = await client.create(query, hooks=call_hooks)
+    print(f"Result: {result3}")
+    
+    # 4. Demonstrate the exact pattern requested by the user
+    print("\n4. User's requested pattern:")
+    print("-" * 50)
+    
+    # This is exactly what the user wanted:
+    call_hooks = Hooks()
+    call_hooks.add_hook(lambda **kwargs: print(f"[USER-PATTERN] {kwargs}"))
+    
+    result4 = await client.create(query, hooks=call_hooks)
+    print(f"Result: {result4}")
+    
+    # 5. Show how both approaches can coexist
+    print("\n5. Both approaches working together:")
+    print("-" * 50)
+    
+    # Traditional approach
+    print("Traditional:")
+    traditional_result = await traditional_task("traditional query", traditional_hooks)
+    
+    # Client approach
+    print("Client-based:")
+    client_result = await client.create("client query")
+    
+    print(f"Both completed successfully!")
+
+
+# Simulate how this might work with Braintrust EvalAsync
+async def simulate_braintrust_integration():
+    """
+    Simulate how the client approach could work alongside EvalAsync.
+    """
+    
+    print("\n" + "=" * 60)
+    print("Simulated Braintrust Integration")
+    print("=" * 60)
+    
+    # Create a client with global hooks
+    client = EvalClient("braintrust-integration-client")
+    client.hooks.add_hook(braintrust_style_hook)
+    
+    # Simulate multiple evaluation queries
+    queries = [
+        "Book a flight to Paris",
+        "Cancel my subscription", 
+        "What are my account settings?"
+    ]
+    
+    for i, query in enumerate(queries, 1):
+        print(f"\nEvaluation {i}:")
+        print("-" * 30)
+        
+        # For some evaluations, add call-specific hooks
+        if i == 2:  # Add special hook for second query
+            call_hooks = Hooks()
+            call_hooks.add_hook(lambda **kwargs: print(f"[SPECIAL] Query {i}: {kwargs.get('input')}"))
+            result = await client.create(query, hooks=call_hooks)
+        else:
+            result = await client.create(query)
+        
+        print(f"Result {i}: {result}")
+
+
+async def main():
+    await demonstrate_integration()
+    await simulate_braintrust_integration()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
\ No newline at end of file