Jupyter Notebook Integration

This example demonstrates how to build a lightweight GenAI testing framework using Corvic MCP and a Jupyter notebook. It reads structured test queries from a CSV file, invokes a Corvic-powered agent, and saves the responses alongside expected answers for comparison.

Use Case

You want to automatically evaluate the output of a Corvic agent across a list of test questions with known expected answers. This is useful for regression testing, QA, and validation of LLM-based applications.

Steps

Configure Corvic Agent Endpoint:
- Set MCP_URL to your deployed Corvic agent’s endpoint.
- Set the HEADERS with your Corvic API token.
Prepare the Input Dataset:
- Create a CSV file with at least the following columns: id, question, expected_answer.
- Set the INPUT_CSV_PATH to the location of this CSV file.
Configure Output:
- Set the OUTPUT_PATH where the agent’s responses will be written as an Excel file.

Input Format (CSV)

id,question,expected_answer
1,"What is the NAICS code for wheat farming?","111140"
2,"How is retail defined in NAICS?","Retail involves selling goods directly to customers..."

Output

An Excel file containing the following columns:

id
question
expected_answer (expected answer)
response (from Corvic)

Code Example

import mcp
import pandas as pd
import asyncio

# Configuration
MCP_URL = "<<MCP_ENDPOINT>>"
CORVIC_API_TOKEN = "<<YOUR_CORVIC_API_TOKEN>>"
INPUT_CSV_PATH = "test_queries.csv"
OUTPUT_PATH = "output.xlsx"

HEADERS = {
    "Authorization": f"Bearer {CORVIC_API_TOKEN}",
    "Content-Type": "application/json"
}

# Load test queries
df = pd.read_csv(INPUT_CSV_PATH)

# Initialize results list
results = []

async def query_agent(question: str):
    """Query the Corvic agent with a question."""
    async with mcp.ClientSession(
        transport=mcp.SSEClientTransport(MCP_URL, headers=HEADERS)
    ) as session:
        await session.initialize()
        
        query_result = await session.call_tool(
            "query",
            arguments={"query_content": question}
        )
        
        response_text = query_result.content[0].text if query_result.content else ""
        return response_text

# Process each query
for idx, row in df.iterrows():
    question_id = row['id']
    question = row['question']
    expected_answer = row['expected_answer']
    
    print(f"Querying ID {question_id}: {question}")
    
    # Query the agent
    response = asyncio.run(query_agent(question))
    
    print(f"Response: {response[:100]}...")
    
    # Store result
    results.append({
        'id': question_id,
        'question': question,
        'expected_answer': expected_answer,
        'response': response
    })

# Save results to Excel
results_df = pd.DataFrame(results)
results_df.to_excel(OUTPUT_PATH, index=False)

print(f"✅ Done. Results saved to {OUTPUT_PATH}")

Example Output

Querying ID 1: What is the NAICS code for wheat farming?
Response: 111140
Querying ID 2: How is retail defined in NAICS?
Response: Retail involves selling goods directly to customers,...
✅ Done. Results saved to /PATH_TO_STORE_RESPONSES/output.xlsx

Resources

Python Integration

Learn more about the Python SDK.

API Integrations