Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -403,4 +403,7 @@ FodyWeavers.xsd

# JetBrains Rider
*.sln.iml
.idea/
.idea/

.DS_Store
**/.venv/
166 changes: 166 additions & 0 deletions samples/DragonCopilot/Workflow/pythonSampleExtension/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,166 @@
## Dragon Copilot Python Sample Extension

A Python FastAPI implementation that mirrors the C# `SampleExtension.Web` for Dragon Copilot. It demonstrates:

- Clinical entity heuristic extraction (vitals, diabetes code, medication concept)
- DSP-style `ProcessResponse` with multiple payload keys
- Adaptive Card visualization via the `adaptive-card` payload
- Comparison tooling against the C# service
- Section-based test payload simulating real clinical note structure

> Disclaimer: This is a learning/sample artifact – not production hardened. Do **not** use with real PHI. Authentication is intentionally disabled for development.

---
## 📚 Contents

- [Dragon Copilot Python Sample Extension](#dragon-copilot-python-sample-extension)
- [1. Features](#1-features)
- [2. Quick Start](#2-quick-start)
- [2.1 Quick Start for Linux and Mac](#21-quick-start-for-linux-and-mac)
- [2.2 Quick Start for Windows](#22-quick-start-for-windows)
- [3. Access the Swagger / OpenAPI](#3-access-the-swagger--openapi)
- [4. Testing APIs with Sample Requests](#4-testing-apis-with-sample-requests)
- [4.1 Testing APIs for Linux / Mac](#41-testing-apis-for-linux--mac)
- [4.2 Testing APIs for Windows](#42-testing-apis-for-windows)
- [5. Response Structure Example](#5-response-structure-example)
- [6. Deploying Your Extension](#6-deploying-your-extension)
- [7. License](#7-license)

---
## 1. Features
**Implemented**
- `/health` & `/v1/health` endpoints
- `/v1/process` returning:
- `sample-entities`
- `adaptive-card`

---
## 2. Quick Start

**Choosing Python 3.12**\
We recommend Python 3.12 over 3.14 in this python sample extension, as a number of ML/AI (especially OSS) Python SDKs do not yet fully support 3.14, which may lead to avoidable integration issues.

**Starting Directory**\
From `pythonSampleExtension` dir:

### 2.1 Quick Start for Linux and Mac
Ensure `python3.12` is installed and can be executed from your cmd shell as `python3.12`.

Run the following cmds in bash/zsh to start server.
```shell
# 1. change to the pythonSampleExtension directory
cd ./samples/DragonCopilot/Workflow/pythonSampleExtension;

# 2. create venv, activate venv and install packages
python3.12 -m venv .venv && source .venv/bin/activate && python3.12 -m pip install --upgrade pip && python3.12 -m pip install -r requirements.txt;

# 3. start server with uvicorn invocation
python3.12 -m uvicorn app.main:app --host 0.0.0.0 --port 5181 --reload
```

### 2.2 Quick Start for Windows
Ensure `python3.12` is installed over Microsoft Store and can be executed from your powershell as `python3.12`.

Run the following cmds in the Powershell to start server.
```powershell
# 1. change to the pythonSampleExtension directory
cd .\samples\DragonCopilot\Workflow\pythonSampleExtension;

# 2. create venv, activate venv and install packages
python3.12 -m venv .venv; . .\.venv\Scripts\Activate.ps1; python3.12 -m pip install --upgrade pip; python3.12 -m pip install -r requirements.txt;

# 3. start server with uvicorn invocation
python3.12 -m uvicorn app.main:app --host 0.0.0.0 --port 5181 --reload
```

## 3 Access the Swagger / OpenAPI
After server start, you shall be able to access the python workflow sample server via Swagger / OpenAPI from your browser with the: `http://localhost:5181/docs`

---
## 4. Testing APIs with Sample Requests
After server started successfully, you can test the python workflow samples from commandline.

### 4.1 Testing APIs for Linux / Mac
**Health API**:
```shell
curl -s http://localhost:5181/health | jq
curl -s http://localhost:5181/v1/health | jq
```

**Process API**:
Minimal process payload:
```shell
curl -s -X POST http://localhost:5181/v1/process \
-H 'Content-Type: application/json' \
-d '{"note":{"resources":[{"content":"Patient has history of diabetes and currently taking metformin. BP recorded."}]}}' | jq
```

Include IDs:
```shell
curl -s -X POST http://localhost:5181/v1/process \
-H 'Content-Type: application/json' \
-H 'x-ms-request-id: demo-req-1' \
-H 'x-ms-correlation-id: demo-corr-1' \
-d '{"note":{"resources":[{"content":"BP 145/98 mmHg; Diabetes risk; taking metformin"}]}}' | jq
```

### 4.2 Testing APIs for Windows
**Health API**:
```powershell
Invoke-RestMethod -Uri "http://localhost:5181/health" | ConvertTo-Json -Depth 5
Invoke-RestMethod -Uri "http://localhost:5181/v1/health" | ConvertTo-Json -Depth 5
```

**Process API**:
Minimal process payload:
```powershell
Invoke-RestMethod -Uri "http://localhost:5181/v1/process" -Method Post -ContentType "application/json" -Body '{"note":{"resources":[{"content":"Patient has history of diabetes and currently taking metformin. BP recorded."}]}}' | ConvertTo-Json -Depth 10
```

Include IDs:
```powershell
Invoke-RestMethod -Uri "http://localhost:5181/v1/process" -Method Post -ContentType "application/json" -Headers @{"x-ms-request-id"="demo-req-1"; "x-ms-correlation-id"="demo-corr-1"} -Body '{"note":{"resources":[{"content":"BP 145/98 mmHg; Diabetes risk; taking metformin"}]}}' | ConvertTo-Json -Depth 10
```

---
## 5. Response Structure Example
You shall see the workflow sample server returns response similar to the following response structure.

```json
{
"success": true,
"message": "Payload processed successfully",
"payload": {
"sample-entities": {
"schema_version": "0.1",
"resources": [ { "type": "ObservationNumber" }, { "type": "MedicalCode" } ]
},
"adaptive-card": {
"schema_version": "0.1",
"resources": [
{
"id": "card1",
"type": "AdaptiveCard",
"subtype": "note",
"adaptive_card_payload": {
"type": "AdaptiveCard",
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"version": "1.6",
"body": []
},
"payloadSources": [],
"dragonCopilotCopyData": "metadata_for_platform"
}
]
}
}
}
```
---
## 6. Deploying Your Extension

The steps above cover running and testing the Python sample extension locally. To fully deploy your extension—including setting up DevTunnels, packaging with the dragon-extension CLI, registering in Azure, uploading to the Dragon Admin Center, and testing inside Dragon Copilot—follow the instructions in the repository root [QUICKSTART.md](../../../../QUICKSTART.md).

---
## 7. License
See root `LICENSE`.
21 changes: 21 additions & 0 deletions samples/DragonCopilot/Workflow/pythonSampleExtension/app/.env
Original file line number Diff line number Diff line change
@@ -0,0 +1,21 @@
# NOTE: These environment variables are placeholders for Azure AI Foundry integration.
# They do NOT use the DGEXT_ prefix required by Settings and are loaded separately
# by future AI agent/client code (not by pydantic-settings).
#
# This file is committed to source control ONLY as a documented example.
# Do NOT put real secrets or production endpoints here.
# Instead, create a local untracked `.env` file (listed in .gitignore) or use
# your environment manager to define these values outside of version control.
#
# PROJECT_ENDPOINT is from the new AI Foundry (project) / Azure AI Service (new version)
# Example:
# PROJECT_ENDPOINT="https://<your-project>.services.ai.azure.com/api/projects/<your-project-name>"
#
# MODEL_DEPLOYMENT_NAME is the name of your deployed model in Azure AI Foundry.
# Example:
# MODEL_DEPLOYMENT_NAME="gpt-4.1-mini"
#
# AGENT_ID is the identifier of the AI agent you want to invoke.
# Example:
# AGENT_ID="asst_xxxxxxx"

Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
"""Python FastAPI sample extension mirroring C# SampleExtension.Web."""
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
model_config = SettingsConfigDict(env_prefix="DGEXT_")

app_name: str = "Dragon Sample Extension (Python)"
version: str = "0.1.0"
# enable_auth: bool = False # Placeholder toggle — not referenced anywhere yet; uncomment when auth middleware is wired up

@lru_cache
def get_settings() -> Settings:
return Settings()
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
from fastapi import FastAPI, Header, HTTPException, Request
# RequestValidationError import kept for reference; handler is commented out below
# from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse, RedirectResponse
from .models import DragonStandardPayload, ProcessResponse
from .service import ProcessingService
from datetime import datetime, timezone
from .config import get_settings
import logging

logger = logging.getLogger("dragon.pyextension")
logging.basicConfig(level=logging.INFO, format="[%(asctime)s] %(levelname)s %(name)s - %(message)s")

settings = get_settings()
app = FastAPI(title=settings.app_name, version=settings.version)
service = ProcessingService()

@app.middleware("http")
async def header_logging_middleware(request: Request, call_next): # basic structured log of tracing headers
req_id = request.headers.get("x-ms-request-id")
corr_id = request.headers.get("x-ms-correlation-id")
logger.info("Incoming %s %s req_id=%s corr_id=%s", request.method, request.url.path, req_id, corr_id)
try:
response = await call_next(request)
except Exception as exc: # noqa: BLE001
logger.exception("Unhandled exception processing request")
return JSONResponse(status_code=500, content={"success": False, "error": "Internal server error"})
return response

@app.get("/", include_in_schema=False)
async def root_redirect():
# Mirror C# swagger root exposure
return RedirectResponse(url="/docs")

@app.get("/health")
async def root_health():
return {"status": "healthy", "version": settings.version}

@app.get("/v1/health")
async def versioned_health():
return {
"service": settings.app_name,
"status": "healthy",
"version": settings.version,
"endpoints": {"process": "/v1/process", "health": "/v1/health"}
}

# @app.exception_handler(RequestValidationError)
# async def validation_exception_handler(request: Request, exc: RequestValidationError): # noqa: D401
# """Return structured details for 422 errors and log them for diagnostics."""
# logger.warning(
# "Validation failed on %s %s errors=%s", request.method, request.url.path, exc.errors()
# )
# return JSONResponse(
# status_code=422,
# content={
# "success": False,
# "message": "Request validation failed",
# "errors": exc.errors(),
# },
# )

@app.post("/v1/process", response_model=ProcessResponse)
async def process_endpoint(
payload: DragonStandardPayload,
x_ms_request_id: str | None = Header(default=None, alias="x-ms-request-id"),
x_ms_correlation_id: str | None = Header(default=None, alias="x-ms-correlation-id"),
):
try:
start_time = datetime.now(timezone.utc)
logger.info("Processing incoming request at %s", start_time)
resp = service.process(payload, x_ms_request_id, x_ms_correlation_id)
elapsed = datetime.now(timezone.utc) - start_time
logger.info("Request processed in %s", elapsed)
return resp
except HTTPException:
raise
except Exception: # noqa: BLE001
logger.exception("Processing failure")
raise HTTPException(status_code=500, detail="Internal server error")
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
from __future__ import annotations
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
from enum import Enum

# Expanded model layer to better mirror the C# sample (not full parity but structurally closer)

class Priority(str, Enum):
High = "High"
Medium = "Medium"
Low = "Low"

class ObservationValue(BaseModel):
text: Optional[str] = None
conceptId: Optional[str] = None

class BaseResource(BaseModel):
id: Optional[str] = None

class MedicalCode(BaseResource):
type: str = Field("MedicalCode", frozen=True)
code: Dict[str, Any] | None = None
priority: Optional[Priority] = None
reason: Optional[str] = None

class ObservationNumber(BaseResource):
type: str = Field("ObservationNumber", frozen=True)
value: Optional[float] = None
valueUnit: Optional[str] = None
priority: Optional[Priority] = None

class ObservationConcept(BaseResource):
type: str = Field("ObservationConcept", frozen=True)
value: Optional[ObservationValue] = None
priority: Optional[Priority] = None

class VisualizationResource(BaseResource):
type: str = Field("AdaptiveCard", frozen=True)
subtype: str | None = None
cardTitle: str | None = None
adaptive_card_payload: Any | None = None
payloadSources: List[Dict[str, Any]] | None = None
dragonCopilotCopyData: str | None = None
partnerLogo: str | None = None
references: List[Dict[str, Any]] | None = None

class NoteResource(BaseModel):
content: Optional[str] = None

## subtype of note shall be lower case 'note'
# _lower_alias is not currently used because the alias config on Note is
# commented out. Uncomment when alias generation is re-enabled.
# def _lower_alias(field_name: str) -> str:
# """Generate lowercase aliases for JSON serialization."""
# return field_name.lower()

class Note(BaseModel):
# Ensure all fields serialize with lowercase keys (even if Python attribute had capitals)
# model_config = ConfigDict(alias_generator=_lower_alias, populate_by_name=True)
# # Explicit type indicator (commonly used in DSP resources) kept lowercase per comment
# type: str = Field(default="note", frozen=True)
document: Dict[str, Any] | None = None
resources: List[NoteResource] | None = None

class SessionData(BaseModel):
sessionId: Optional[str] = None

class DspResponse(BaseModel):
schema_version: str | None = None
document: Dict[str, Any] | None = None
resources: List[Any] = Field(default_factory=list)

class DragonStandardPayload(BaseModel):
note: Optional[Note] = None
sessionData: SessionData | None = None

class ProcessResponse(BaseModel):
success: bool = False
message: Optional[str] = None
payload: Dict[str, DspResponse | Any] = Field(default_factory=dict)
Loading
Loading